4. Kinesis Data Streams实时接入设计

Kinesis Data Streams（简称 KDS）是 AWS 上承载实时数据接入的核心服务，它把持续到达的事件按顺序写入一个可水平扩展的流中，供下游消费者在秒级内读取。在批流一体平台里，KDS 通常扮演“实时入口”的角色：业务系统、日志采集器、IoT 设备或 CDC 组件把数据推入 KDS，Flink、Lambda、Kinesis Data Analytics 或自建消费者再从流中拉取并写入数据湖、数仓或下游服务。本节围绕四个关键动作展开：创建数据流、配置分片与保留周期、编写生产者与消费者示例、验证低延迟写入与读取。

一、数据流、分片与保留周期的基本关系

创建数据流时，最需要理解的是“分片（Shard）”这一概念。分片是 KDS 的吞吐与并行基本单位，每个分片提供固定的写入与读取能力：写入上限为 1 MB/s 或 1000 条记录/秒，读取上限为 2 MB/s（共享读取模式下所有消费者合计）。数据流的总容量等于分片数乘以单分片容量，因此分片数直接决定了流的写入吞吐上限和消费者并行度。

每条记录写入时必须指定一个分区键（Partition Key），KDS 对分区键做哈希后映射到某个分片。相同分区键的记录会进入同一分片，从而保证该键上的顺序性；不同分区键则可能分散到不同分片，实现负载均衡。如果分区键分布不均，就会出现“热分片”，部分分片被打满而其他分片空闲，这是实时接入最常见的性能问题之一。

保留周期（Retention Period）决定记录在流中可被读取的时间窗口，默认 24 小时，可调整为 24 小时至 8760 小时（365 天）。保留周期越长，消费者重放历史数据的能力越强，但存储成本也越高。对于需要故障重放或回溯计算的批流一体场景，通常会把保留周期设为 48 小时到 7 天，而不是停留在默认值。

二、创建数据流与配置分片

创建数据流可以通过控制台、AWS CLI 或 SDK 完成。以 CLI 为例，创建一个名为 realtime-ingest-demo、3 个分片、保留周期 48 小时的流：

aws kinesis create-stream \
  --stream-name realtime-ingest-demo \
  --shard-count 3

aws kinesis increase-stream-retention-period \
  --stream-name realtime-ingest-demo \
  --retention-period-hours 48

aws kinesis describe-stream-summary \
  --stream-name realtime-ingest-demo

分片数的估算方法是：先统计峰值写入速率。假设业务峰值每秒产生 800 条记录、平均每条 1.2 KB，则写入带宽约为 800 × 1.2 KB ≈ 960 KB/s，记录数 800 条/s。按单分片 1 MB/s 和 1000 条/s 计算，1 个分片在带宽上接近上限、在记录数上接近上限，因此至少需要 2 个分片才能留出余量；考虑到后续增长和消费者并行度，示例中选择 3 个分片。如果使用按需模式（On-Demand），则无需预先指定分片数，KDS 会根据流量自动扩缩，适合流量波动大或难以预估的场景。

保留周期建议在创建后立即调整。默认 24 小时对于需要重放或故障恢复的批流一体场景往往不够，示例中设为 48 小时，兼顾成本与重放能力。若下游 Flink 作业需要从更早位点回溯，可进一步延长到 168 小时（7 天）。

三、编写生产者示例

生产者使用 AWS SDK 的 put_record 或 put_records 写入。单条写入适合低吞吐、强顺序场景；批量写入（最多 500 条或 5 MB）适合高吞吐场景。下面是一个 Python 生产者示例，模拟订单事件写入，分区键使用 user_id，保证同一用户的事件进入同一分片：

import boto3, json, time, uuid

kinesis = boto3.client('kinesis', region_name='ap-northeast-1')
STREAM = 'realtime-ingest-demo'

def put_order(user_id, amount):
    record = {
        'order_id': str(uuid.uuid4()),
        'user_id': user_id,
        'amount': amount,
        'ts': int(time.time() * 1000)
    }
    resp = kinesis.put_record(
        StreamName=STREAM,
        Data=json.dumps(record).encode('utf-8'),
        PartitionKey=user_id
    )
    return resp['ShardId'], resp['SequenceNumber']

for i in range(10):
    shard, seq = put_order(f'user-{i % 3}', 100 + i)
    print(f'写入分片={shard} 序列号={seq}')

运行后可以看到，user-0、user-1、user-2 分别被哈希到不同分片，输出中会显示对应的 ShardId 和 SequenceNumber。SequenceNumber 是 KDS 为每条记录分配的唯一递增标识，消费者用它记录位点。

四、编写消费者示例

消费者通过 get_shard_iterator 获取迭代器，再用 get_records 循环拉取。迭代器类型常用 LATEST（从最新开始）、TRIM_HORIZON（从最早可用开始）和 AT_SEQUENCE_NUMBER（从指定位点开始）。下面示例从 TRIM_HORIZON 读取，并打印每条记录的分区和数据：

import boto3, json

kinesis = boto3.client('kinesis', region_name='ap-northeast-1')
STREAM = 'realtime-ingest-demo'

shards = kinesis.list_shards(StreamName=STREAM)['Shards']
for shard in shards:
    sid = shard['ShardId']
    it = kinesis.get_shard_iterator(
        StreamName=STREAM,
        ShardId=sid,
        ShardIteratorType='TRIM_HORIZON'
    )['ShardIterator']

    while it:
        out = kinesis.get_records(ShardIterator=it, Limit=100)
        for rec in out['Records']:
            data = json.loads(rec['Data'])
            print(f"[{sid}] user={data['user_id']} amount={data['amount']} "
                  f"seq={rec['SequenceNumber']}")
        it = out.get('NextShardIterator')

生产环境中，消费者需要把处理过的 SequenceNumber 持久化（如写入 DynamoDB 或 Redis），以便重启后从断点继续，而不是每次从 TRIM_HORIZON 重放。KCL（Kinesis Client Library）已经内置了分片租约和检查点机制，适合多实例并行消费；自建消费者则需要自行处理分片分配与位点保存。

五、验证低延迟写入与读取

验证的核心是观察“写入时间”与“读取时间”的差值，以及消费者落后程度。可以在生产者记录中写入事件时间戳 ts，消费者读取时计算 now - ts，得到端到端时延。示例中若生产者每秒写入 10 条、消费者每秒拉取一次，正常情况下端到端时延应在 1～3 秒内。

同时应关注 CloudWatch 指标：PutRecord.Success 和 PutRecord.Latency 反映写入健康度；GetRecords.IteratorAgeMilliseconds（即 MillisBehindLatest）反映消费者落后时间，稳定在 2 秒以内说明消费跟得上；WriteProvisionedThroughputExceeded 若持续大于 0，说明分片写入被打满，需要增加分片或优化分区键分布。若发现某个分片的 IteratorAge 明显高于其他分片，通常是分区键倾斜导致的热分片，应检查分区键基数并考虑加入随机后缀打散。