ConsumeKinesisStream

文档中心 › 组件索引 › ConsumeKinesisStream

ConsumeKinesisStream Processor

类型全限定名：cn.org.acal.lynxus.processors.aws.kinesis.stream.ConsumeKinesisStream

所属 Bundle：cn.org.acal.lynxus:lynxus-aws-nar:2.11.8

描述

从指定的 AWS Kinesis 流读取数据，并为每个处理的记录（原始）输出一个 FlowFile，或者如果配置了记录读取器和记录写入器，则输出一批处理记录的 FlowFile。

标签

amazonawskinesisconsumestream

属性

下表中，加粗的属性为必填项；其余为可选项。表格同时标注默认值、是否支持表达式语言，以及是否为敏感属性（值会被加密）。

名称默认值可选值描述
Amazon Kinesis Stream NameKinesis 流的名称
Application NameKinesis 流读取器应用程序名称。
Record Reader控制器服务 API：
RecordReaderFactory
cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8		用于读取接收消息的 Record Reader。可以通过表达式语言 '${kinesis.name}' 引用 Kinesis 流名称来访问 Schema。如果未指定 Record Reader/Writer，则每条 Kinesis 记录将创建一个 FlowFile。
Record Writer控制器服务 API：
RecordSetWriterFactory
cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8		用于将记录序列化到输出 FlowFile 的 Record Writer。可以通过表达式语言 '${kinesis.name}' 引用 Kinesis 流名称来访问 Schema。如果未指定 Record Reader/Writer，则每条 Kinesis 记录将创建一个 FlowFile。
Output StrategyUSE_VALUE
  • Use Content as Value — Write only the Kinesis Record value to the FlowFile record.
  • Use Wrapper — 将 Kinesis Record 值和元数据写入 FlowFile 记录。
用于将 Kinesis 记录输出到 FlowFile 记录的格式。
FlowFile Handling On Schema DifferenceCREATE_FLOW_FILE
  • Create FlowFile — 为每条具有不同 schema 的记录创建一个新的 FlowFile。之前的 FlowFile 将以具有先前 schema 的记录完成。 发出的 FlowFile 将包含连续的记录序列。
  • Group Records By Schema — 将具有相同 schema 的记录分组到单个 FlowFile 中。如果遇到具有不同 schema 的记录，将为 新 schema 创建一个新的 FlowFile。发出的 FlowFile 可能包含非连续的记录。当 schema 频繁变化且需要最高性能时，此策略很有用。
当 Kinesis Stream 中的记录在单个批次中更改其架构时使用的策略。
Regionus-west-2
  • Africa (Cape Town) — AWS Region Code : af-south-1
  • Asia Pacific (Hong Kong) — AWS Region Code : ap-east-1
  • Asia Pacific (Hyderabad) — AWS Region Code : ap-south-2
  • Asia Pacific (Jakarta) — AWS Region Code : ap-southeast-3
  • Asia Pacific (Malaysia) — AWS Region Code : ap-southeast-5
  • Asia Pacific (Melbourne) — AWS Region Code : ap-southeast-4
  • Asia Pacific (Mumbai) — AWS Region Code : ap-south-1
  • Asia Pacific (New Zealand) — AWS Region Code : ap-southeast-6
  • Asia Pacific (Osaka) — AWS Region Code : ap-northeast-3
  • Asia Pacific (Seoul) — AWS Region Code : ap-northeast-2
  • Asia Pacific (Singapore) — AWS Region Code : ap-southeast-1
  • Asia Pacific (Sydney) — AWS Region Code : ap-southeast-2
  • Asia Pacific (Taipei) — AWS Region Code : ap-east-2
  • Asia Pacific (Thailand) — AWS Region Code : ap-southeast-7
  • Asia Pacific (Tokyo) — AWS Region Code : ap-northeast-1
  • AWS European Sovereign Cloud (Germany) — AWS Region Code : eusc-de-east-1
  • aws global region — AWS Region Code : aws-global
  • AWS GovCloud (US-East) — AWS Region Code : us-gov-east-1
  • AWS GovCloud (US-West) — AWS Region Code : us-gov-west-1
  • aws-cn global region — AWS Region Code : aws-cn-global
  • aws-iso global region — AWS Region Code : aws-iso-global
  • aws-iso-b global region — AWS Region Code : aws-iso-b-global
  • aws-iso-e global region — AWS Region Code : aws-iso-e-global
  • aws-iso-f global region — AWS Region Code : aws-iso-f-global
  • aws-us-gov global region — AWS Region Code : aws-us-gov-global
  • Canada (Central) — AWS Region Code : ca-central-1
  • Canada West (Calgary) — AWS Region Code : ca-west-1
  • China (Beijing) — AWS Region Code : cn-north-1
  • China (Ningxia) — AWS Region Code : cn-northwest-1
  • EU ISOE West — AWS Region Code : eu-isoe-west-1
  • Europe (Frankfurt) — AWS Region Code : eu-central-1
  • Europe (Ireland) — AWS Region Code : eu-west-1
  • Europe (London) — AWS Region Code : eu-west-2
  • Europe (Milan) — AWS Region Code : eu-south-1
  • Europe (Paris) — AWS Region Code : eu-west-3
  • Europe (Spain) — AWS Region Code : eu-south-2
  • Europe (Stockholm) — AWS Region Code : eu-north-1
  • Europe (Zurich) — AWS Region Code : eu-central-2
  • Israel (Tel Aviv) — AWS Region Code : il-central-1
  • Mexico (Central) — AWS Region Code : mx-central-1
  • Middle East (Bahrain) — AWS Region Code : me-south-1
  • Middle East (UAE) — AWS Region Code : me-central-1
  • South America (Sao Paulo) — AWS Region Code : sa-east-1
  • US East (N. Virginia) — AWS Region Code : us-east-1
  • US East (Ohio) — AWS Region Code : us-east-2
  • US ISO East — AWS Region Code : us-iso-east-1
  • US ISO WEST — AWS Region Code : us-iso-west-1
  • US ISOB East (Ohio) — AWS Region Code : us-isob-east-1
  • US ISOB West — AWS Region Code : us-isob-west-1
  • US ISOF EAST — AWS Region Code : us-isof-east-1
  • US ISOF SOUTH — AWS Region Code : us-isof-south-1
  • US West (N. California) — AWS Region Code : us-west-1
  • US West (Oregon) — AWS Region Code : us-west-2
  • Use Custom Region — 用户定义的区域，可以是非 AWS 区域，和/或来自表达式语言（Expression Language）
服务所在的 AWS 区域
Custom Region自定义区域，例如 AWS 兼容服务提供商的区域
支持表达式语言：是（仅使用环境/系统变量求值）
Endpoint Override URL用于替代 AWS 默认端点的端点 URL，包括协议、主机、端口和路径。AWS 库根据 AWS 区域选择端点 URL，但此属性会覆盖所选端点 URL，允许与其他 S3 兼容端点一起使用。
支持表达式语言：是（仅使用环境/系统变量求值）
DynamoDB OverrideDynamoDB 覆盖以使用非 AWS 部署
支持表达式语言：是（仅使用环境/系统变量求值）
Initial Stream PositionLATEST
  • LATEST — 从分片中最新记录之后的位置开始读取，以便始终读取分片中最新的数据。
  • TRIM_HORIZON — 从系统中分片内最后一条未修剪的记录开始读取，即分片中最旧的数据记录。
  • AT_TIMESTAMP — 从由特定时间戳（在 Timestamp 值中提供）所表示的位置开始读取。
读取 Kinesis 流的初始位置。
流位置时间戳开始读取 Kinesis 记录的流中时间戳位置。如果读取 Kinesis 流的初始位置。为AT_TIMESTAMP，则此项为必填。使用时间戳格式将值解析为日期。
时间戳格式yyyy-MM-dd HH:mm:ss用于将流位置时间戳解析为日期并将 Kinesis 记录的近似到达时间戳转换为 FlowFile 属性的格式。
支持表达式语言：是（仅使用环境/系统变量求值）
Failover Timeout30 secsKinesis 客户端库故障转移超时
Graceful Shutdown Timeout20 secsKinesis 客户端库正常关闭超时
Checkpoint Interval3 secsKinesis 检查点之间的间隔
Retry Count10重试 Kinesis 操作的次数（进程记录、检查点、关闭）
Retry Wait1 secKinesis 操作重试之间的间隔（进程记录、检查点、关闭）
Report Metrics to CloudWatchfalse
  • true
  • false
是否向 CloudWatch 报告 Kinesis 使用指标。
Communications Timeout30 secs
AWS Credentials Provider Service控制器服务 API：
AwsCredentialsProviderService
cn.org.acal.lynxus-lynxus-aws-service-api-nar-2.11.8		用于获取AWS凭证提供者的Controller Service
Proxy Configuration Service控制器服务 API：
ProxyConfigurationService
cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8		指定代理配置 Controller Service 来代理网络请求。

动态属性

动态属性允许用户自定义属性的名称与值。

名称值描述
Kinesis Client Library (KCL) 配置属性名称要在 KCL 配置属性中设置的值使用必需值覆盖默认 KCL 配置 ConfigsBuilder 属性。支持直接在 ConfigsBuilder 上设置值（如 'namespace'），以及嵌套构建器上的属性。例如，要设置 configsBuilder.retrievalConfig().maxListShardsRetryAttempts(value)，将属性命名为 'retrievalConfig.maxListShardsRetryAttempts'。仅支持设置简单属性值，如 String、int、long 和 boolean。不允许覆盖由非动态处理器属性处理的 KCL 配置设置。

关系 (Relationships)

名称描述自动终止
successFlowFiles被路由到成功关系否

写入的 FlowFile 属性

名称描述
aws.kinesis.partition.key从分片读取的（最后一个）Kinesis 记录的分区键
aws.kinesis.shard.id从中读取 Kinesis 记录的分片 ID
aws.kinesis.sequence.number其分片中（最后一个）Kinesis 记录的唯一标识符
aws.kinesis.approximate.arrival.timestamp从流中读取的（最后一个）Kinesis Record 的大致到达时间戳
mime.type将 mime.type 属性设置为Record Writer指定的 MIME 类型（如果已配置）
record.count由Record Writer写入FlowFile的记录数（如果配置）
record.error.message此属性在失败时提供Record Reader或Record Writer遇到的错误消息（如果已配置）

状态管理

该组件不存储状态。

受限 (Restricted)

该组件不受限。

输入要求

该组件不允许有传入连接。

系统资源考量

资源描述
CPUKinesis Client Library 用于创建 Worker 线程以消费 Kinesis 记录。Worker 在此处理器被触发时初始化并启动。它持续运行，根据需要生成 Kinesis Record Processor 来获取 Kinesis 记录。Worker 线程（及任何子 Record Processor 线程）不受 Lynxus 常规调度器作为并发线程池的一部分控制，并且在此处理器停止之前不会释放。
NETWORKKinesis Client Library 将持续轮询新记录，每次调用最多请求最大数量的记录/字节。这可能导致持续的网络使用。

另请参阅

详细说明

ConsumeKinesisStream

流式处理与批处理

ConsumeKinesisStream 会获取所配置 Kinesis Stream 中遇到的所有 Kinesis Record。常见的用例大致分为两类。

逐消息用例

默认情况下，该 Processor 会为 Stream 中的每条 Kinesis Record（消息） 创建一个独立的 FlowFile，并添加 shard id、sequence number 等 Attribute。

逐批次用例

另一个常见用例是希望把从 Stream 获取的所有 Kinesis Record 作为单个 FlowFile 按批处理。

ConsumeKinesisStream Processor 可以可选地配置 Record Reader 和 Record Writer。配置后，只会创建单个 FlowFile，其中为 Kinesis Record （消息）批次内的每条 Record 包含一条记录，而不是每条 Kinesis Record 一个 FlowFile。

此模式下输出的 FlowFile 会包含标准的 record.* Attribute，以及相同的 Kinesis Shard ID、Sequence Number 和 Approximate Arrival Timestamp； 但这些值对应的是构成该 FlowFile 内容的消息批次中最后一条被处理的 Kinesis Record。

设置 Record Writer 后，Output Strategy 可以设为 Use Wrapper 或 Use Content。选择 Use Wrapper 时，Kinesis Record 的原始内容会被 包裹在 value 键下，并额外填充 metadata 键，其中包含 Stream Name、 Shard ID、Partition Key、Sequence Number、Sub Sequence Number、 Approximate Arrival Timestamp，以及由 Sequence 和 Sub Sequence Number 拼接而成的 Sharded Sequence Number，便于通过单次比较确定记录顺序。 选择 Use Content 时，Kinesis Record 的原始内容会原样使用。

© 2026 ACAL Organization. Licensed under Apache License 2.0.