📚 课程目录第4章流式数据接入

4. 流式数据接入

在批流一体的数据管道中，流式接入层承担着把持续产生的数据"接住、缓冲、分发"的职责。AWS 提供了两个定位不同的服务：Kinesis Data Streams（KDS）是低延迟、可重放、支持多消费者的实时流存储；Data Firehose则是托管式投递服务，负责把数据自动批量写入 S3、Redshift、OpenSearch 等目标。理解两者的分工，是设计接入层的第一步。

一、KDS 与 Firehose 的定位差异

KDS 的核心抽象是分片（Shard）。每个分片是一条有序的追加日志，写入端按分区键把记录路由到某个分片，读取端按分片顺序消费。分片决定了吞吐上限：单分片写入 1 MB/s 或 1000 条记录/s，读取 2 MB/s。数据默认保留 24 小时，可延长到 365 天，因此支持"重放"——同一份数据可以被多个消费者独立读取，这正是实时计算、审计、归档可以并行消费的基础。

Firehose 没有分片概念，用户只需指定投递目标、缓冲大小与缓冲间隔，服务会自动扩缩容。它适合"采集后直接落库/落湖"的场景，但不支持多消费者重放，延迟通常在 60 秒以上（取决于缓冲策略）。因此常见架构是：KDS 作为实时处理与多路消费的入口，Firehose 作为 KDS 的下游消费者之一，把原始数据归档到 S3。

二、创建 KDS 与分片规划

创建流时最关键的两个参数是分片数量和数据保留期。分片数按峰值写入量估算：若业务峰值写入为 8 MB/s，则至少需要 8 个分片（8 ÷ 1 MB/s）。若同时要求 1000 条/s 的记录数上限，则取两种估算的较大值。例如峰值 8 MB/s 且 12000 条/s，则需 max(8, 12) = 12 个分片。

aws kinesis create-stream \
  --stream-name orders-stream \
  --shard-count 12 \
  --stream-mode-details StreamMode=PROVISIONED

# 查看分片与哈希键范围
aws kinesis describe-stream --stream-name orders-stream \
  --query "StreamDescription.Shards[].{Id:ShardId,Hash:HashKeyRange}"

分片模式有两种：PROVISIONED 需手动管理分片数，成本可预测；ON_DEMAND 由服务自动扩缩，适合流量波动大、难以预估的场景，但单价更高。生产环境建议先用 ON_DEMAND 观察真实流量，再切换到 PROVISIONED 优化成本。

三、分片管理与再平衡

分片不是一成不变的。当某个分片持续触发 WriteProvisionedThroughputExceeded 指标时，说明写入超过 1 MB/s，需要 Split Shard 把一个分片拆成两个；当多个分片长期利用率低于 30% 时，可用 Merge Shard 合并相邻分片以节省成本。注意：拆分/合并会改变分片数量，进而改变 hash(PK) % N 的映射结果，因此分区键的分布均匀性比分区键本身更重要。若大量记录使用同一个分区键（如固定值 "default"），所有数据会挤进一个分片，扩容也无法缓解。

分区键设计原则：选择基数高、分布均匀的字段，如用户 ID、设备 ID、订单号；避免使用时间戳、状态码等低基数字段。

四、生产者实现：KPL 与 PutRecord

生产者有两种典型方式。简单场景用 PutRecord / PutRecords（批量最多 500 条、5 MB）；高吞吐场景推荐 Kinesis Producer Library（KPL），它自动做聚合（Aggregation）、重试和分片映射，能把多条小记录打包成一条 Kinesis 记录，显著降低单位成本。

import boto3, json
kinesis = boto3.client("kinesis", region_name="us-east-1")

def put_order(order):
    resp = kinesis.put_record(
        StreamName="orders-stream",
        Data=json.dumps(order).encode("utf-8"),
        PartitionKey=order["user_id"]   # 高基数，保证均匀分布
    )
    return resp["ShardId"], resp["SequenceNumber"]

# 批量写入示例
records = [{"Data": json.dumps(o).encode(), "PartitionKey": o["user_id"]}
           for o in batch]
kinesis.put_records(StreamName="orders-stream", Records=records)

批量写入时需检查返回的 FailedRecordCount，对失败记录做指数退避重试。若返回 ProvisionedThroughputExceededException，说明分片写入已达上限，应降低发送速率或扩容。

五、消费者实现：KCL 与 Lambda

消费者侧，Kinesis Client Library（KCL）是最常用的方案。它基于 DynamoDB 表记录每个分片的处理位点（Checkpoint），并自动做分片租约（Lease）管理：当消费者实例增减时，KCL 会重新分配分片，保证一个分片同一时刻只被一个 worker 处理。每个分片对应一个 ShardRecordProcessor，在 processRecords 中处理记录，处理成功后调用 checkpoint() 提交位点。

# Lambda 消费 KDS 的关键配置
EventSourceMapping:
  EventSourceArn: arn:aws:kinesis:us-east-1:123456789012:stream/orders-stream
  StartingPosition: LATEST        # 或 TRIM_HORIZON 从头消费
  BatchSize: 100
  MaximumBatchingWindowInSeconds: 5
  ParallelizationFactor: 2        # 每个分片并发 2 个 Lambda
  BisectBatchOnFunctionError: true
  MaximumRetryAttempts: 3
  DestinationConfig:
    OnFailure:
      Destination: arn:aws:sqs:...:orders-dlq

Lambda 消费 KDS 时，ParallelizationFactor 允许单个分片并发多个 Lambda 实例，但会牺牲部分顺序性；BisectBatchOnFunctionError 在批处理失败时把批次一分为二重试，便于定位坏记录；失败记录最终进入 SQS 死信队列，避免阻塞整个分片。

六、Firehose 投递配置

Firehose 的创建重点是缓冲策略与目标格式。缓冲大小（1–128 MB）和缓冲间隔（60–900 秒）共同决定投递时机，二者先到先触发。若目标是 S3，可开启 Dynamic Partitioning 按字段（如 year/month/day/hour）自动分区，避免小文件过多。

aws firehose create-delivery-stream \
  --delivery-stream-name orders-to-s3 \
  --delivery-stream-type KinesisStreamAsSource \
  --kinesis-stream-source-configuration \
      KinesisStreamARN=arn:aws:kinesis:...:stream/orders-stream,RoleARN=arn:aws:iam::...:role/firehose-role \
  --s3-destination-configuration \
      RoleARN=arn:aws:iam::...:role/firehose-role,\
BucketARN=arn:aws:s3:::orders-lake,\
BufferingHints={SizeInMBs=64,IntervalInSeconds=300},\
Prefix=raw/year=!{partitionKeyFromQuery:year}/month=!{partitionKeyFromQuery:month}/

Firehose 写入 S3 时默认按投递时间生成文件名，若下游需要按事件时间分区，必须启用 Dynamic Partitioning 并在源数据中携带时间字段，否则会出现"数据落在错误日期目录"的问题。

七、如何判断接入层是否健康

接入层的健康度主要看四个指标：IncomingRecords / IncomingBytes（写入速率是否接近分片上限）、WriteProvisionedThroughputExceeded（是否触发限流）、IteratorAgeMilliseconds（消费者滞后程度，持续增长说明消费跟不上生产）、DeliveryToS3.Success（Firehose 投递成功率）。当 IteratorAge 持续上升时，优先检查消费者处理逻辑是否变慢，其次考虑增加分片或提高 ParallelizationFactor。

整体而言，KDS 负责"接得住、可重放、多路消费"，Firehose 负责"自动投递、免运维落湖"。把两者组合使用，就能构建一条既能实时处理、又能可靠归档的流式接入链路。