ConsumeKinesis

ConsumeKinesis 所需的 IAM 权限

必须为 AWS Credentials Provider Service 中配置的 IAM 角色或用户添加 以下权限。

仅当 Metrics Publishing 设为 CloudWatch 时才需要 CloudWatch 权限。

以下是 ConsumeKinesis 的示例策略文档。

{
    "Version":"2012-10-17",		 	 	 
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kinesis:DescribeStream",
                "kinesis:DescribeStreamSummary",
                "kinesis:RegisterStreamConsumer",
                "kinesis:GetRecords",
                "kinesis:GetShardIterator",
                "kinesis:ListShards"
            ],
            "Resource": [
                "arn:aws:kinesis:{Region}:{Account}:stream/{Stream Name}"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "kinesis:SubscribeToShard",
                "kinesis:DescribeStreamConsumer"
            ],
            "Resource": [
                "arn:aws:kinesis:{Region}:{Account}:stream/{Stream Name}/consumer/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "dynamodb:CreateTable",
                "dynamodb:DescribeTable",
                "dynamodb:UpdateTable",
                "dynamodb:GetItem",
                "dynamodb:UpdateItem",
                "dynamodb:PutItem",
                "dynamodb:DeleteItem",
                "dynamodb:Scan"
            ],
            "Resource": [
                "arn:aws:dynamodb:{Region}:{Account}:table/{Application Name}"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "dynamodb:CreateTable",
                "dynamodb:DescribeTable",
                "dynamodb:GetItem",
                "dynamodb:UpdateItem",
                "dynamodb:PutItem",
                "dynamodb:DeleteItem",
                "dynamodb:Scan"
            ],
            "Resource": [
                "arn:aws:dynamodb:{Region}:{Account}:table/{Application Name}-WorkerMetricStats",
                "arn:aws:dynamodb:{Region}:{Account}:table/{Application Name}-CoordinatorState"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "dynamodb:Query"
            ],
            "Resource": [
                "arn:aws:dynamodb:{Region}:{Account}:table/{Application Name}/index/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "cloudwatch:PutMetricData"
            ],
            "Resource": "*"
        }
    ]
}

注意： 请把 ARN 中的 "{Region}"、"{Account}"、"{Stream Name}" 和 "{Application Name}" 分别替换为你的 AWS 区域、AWS 账号、Kinesis 数据流 名称，以及 ConsumeKinesis 的 Application Name 属性。

消费者类型

不同消费者类型的对比（引自 Amazon Kinesis Streams 文档）：

记录处理

当 Processing Strategy 属性设为 RECORD 时，ConsumeKinesis 以 Record 模式运行。该模式下，Processor 使用配置的 Record Reader 从 Kinesis 流读取记录，并使用配置的 Record Writer 写入 FlowFile。

Processor 会尝试把相同 schema 的多条记录批量放入单个 FlowFile， 以优化创建的 FlowFile 数量。

Schema 变更

ConsumeKinesis 支持动态变化的 Kinesis 记录 schema。当遇到带有新 schema 的记录时，当前打开的 FlowFile 会被关闭，并为新 schema 创建 新的 FlowFile。得益于此，即使记录 schema 发生变化，该 Processor 也能保持单个 Shard 内的记录顺序。

请注意，该 Processor 依赖 Record Reader 为每条记录提供正确的 schema。使用带 schema 推断的 Reader 可能产生大量不同的 schema， 导致过度创建 FlowFile。如果关注性能，建议使用带预定义 schema 或 schema registry 的 Reader。

输出策略

该 Processor 通过 Output Strategy 属性提供多种把 Kinesis 记录 转换为 FlowFile 的策略：

• Use Content as Value（默认）只把 Kinesis 记录的值写入 FlowFile。
• Use Wrapper 把 Kinesis Record 的值和元数据写入 FlowFile 记录的独立字段。
• Inject Metadata 把 Kinesis Record 的值写入 FlowFile 记录，并为其添加一个包含元数据的子记录。

写入的元数据包含以下字段：

• stream: 记录所来源的 Kinesis 流的名称。
• shardId: 记录所来源分片的标识符。
• sequenceNumber: 记录的序列号。
• subSequenceNumber: 记录的子序列号，在多个较小记录聚合为单个 Kinesis 记录时使用。如果记录不属于批次，该值为 0。
• shardedSequenceNumber: 序列号和子序列号的组合。可用于在分片内唯一标识一条记录。
• partitionKey: 记录的分区键。
• approximateArrival: 记录的近似到达时间戳（自纪元以来的毫秒数）。

选择 Use Wrapper 时使用的记录 schema 如下（Avro 格式）：

{
  "type": "record",
  "name": "lynxusRecord",
  "namespace": "cn.org.acal.lynxus",
  "fields": [
    {
      "name": "value",
      "type": [
        {
          < 由 Record Reader 为 Kinesis 消息确定的字段 >
        },
        "null"
      ]
    },
    {
      "name": "kinesisMetadata",
      "type": [
        {
          "type": "record",
          "name": "metadataType",
          "fields": [
            { "name": "stream", "type": ["string", "null"] },
            { "name": "shardId", "type": ["string", "null"] },
            { "name": "sequenceNumber", "type": ["string", "null"] },
            { "name": "subSequenceNumber", "type": ["long", "null"] },
            { "name": "shardedSequenceNumber", "type": ["string", "null"] },
            { "name": "partitionKey", "type": ["string", "null"] },
            { "name": "approximateArrival", "type": [ { "type": "long", "logicalType": "timestamp-millis" }, "null" ] }
          ]
        },
        "null"
      ]
    }
  ]
}

选择 "Inject Metadata" 时使用的记录 schema 如下（Avro 格式）：

{
  "type": "record",
  "name": "lynxusRecord",
  "namespace": "cn.org.acal.lynxus",
  "fields": [
    < 由 Record Reader 为 Kinesis 消息确定的字段 >,
    {
      "name": "kinesisMetadata",
      "type": [
        {
          "type": "record",
          "name": "metadataType",
          "fields": [
            { "name": "stream", "type": ["string", "null"] },
            { "name": "shardId", "type": ["string", "null"] },
            { "name": "sequenceNumber", "type": ["string", "null"] },
            { "name": "subSequenceNumber", "type": ["long", "null"] },
            { "name": "shardedSequenceNumber", "type": ["string", "null"] },
            { "name": "partitionKey", "type": ["string", "null"] },
            { "name": "approximateArrival", "type": [ { "type": "long", "logicalType": "timestamp-millis" }, "null" ] }
          ]
        },
        "null"
      ]
    }
  ]
}

选择 Use Wrapper 策略时，JsonRecordSetWriter 输出的 FlowFile 内容示例：

[
  {
    "value": {
      "address": "1234 First Street",
      "zip": "12345",
      "account": {
        "name": "Acme",
        "number": "AC1234"
      }
    },
    "kinesisMetadata" : {
      "stream" : "stream-name",
      "shardId" : "shardId-000000000000",
      "sequenceNumber" : "123456789",
      "subSequenceNumber" : 3,
      "shardedSequenceNumber" : "12345678900000000000000000003",
      "partitionKey" : "123",
      "approximateArrival" : 1756459596788
    }
  }
]

选择 Inject Metadata 策略时，JsonRecordSetWriter 输出的 FlowFile 内容示例：

[
  {
    "address": "1234 First Street",
    "zip": "12345",
    "account": {
      "name": "Acme",
      "number": "AC1234"
    },
    "kinesisMetadata" : {
      "stream" : "stream-name",
      "shardId" : "shardId-000000000000",
      "sequenceNumber" : "123456789",
      "subSequenceNumber" : 3,
      "shardedSequenceNumber" : "12345678900000000000000000003",
      "partitionKey" : "123",
      "approximateArrival" : 1756459596788
    }
  }
]