4. Kinesis Data Streams生产端实战

在实时数据分析系统中，生产端是整条数据链路的起点。Kinesis Data Streams 的写入可靠性、吞吐能力和延迟表现，很大程度上取决于生产者如何选择写入方式、如何组织分区键、如何处理失败重试以及如何应对背压。本节围绕三种典型生产端实现方式展开：KPL（Kinesis Producer Library）、Kinesis Agent 和 AWS SDK，并结合重试与背压处理给出可落地的工程实践。

三种生产端的定位与选择

KPL 是 AWS 提供的高层封装库，适合需要高吞吐、自动聚合、异步写入和内置重试的场景。它会把多条小记录聚合为一条 Kinesis PutRecords 请求，从而显著降低单条记录的单位写入成本。Kinesis Agent 是一个独立运行的代理进程，适合把服务器上的日志文件、文本文件持续采集并投递到 Kinesis，通常用于 EC2 或本地服务器的日志采集场景。AWS SDK 则是最底层的编程接口，适合需要精细控制分区键、重试策略、超时和错误处理的定制化生产者。

选择时可以参考以下判断：如果生产端是应用程序且追求高吞吐，优先考虑 KPL；如果是日志文件采集且希望少写代码，优先考虑 Kinesis Agent；如果需要与业务逻辑深度耦合、精确控制每条记录的写入行为，则使用 AWS SDK。三者并非互斥，实际系统中常混合使用。

使用 KPL 构建高吞吐生产者

KPL 的核心优势在于聚合与异步。默认情况下，KPL 会在本地缓冲记录，当达到 RecordMaxBufferedTime（默认 100ms）或 AggregationMaxCount（默认 4294967295，实际受请求大小限制）时触发聚合发送。聚合后的记录以 PutRecords 方式写入，单次请求最多 500 条记录、总大小不超过 5MB。这意味着 KPL 能显著减少 API 调用次数，但也会引入额外延迟。

一个典型的 KPL 配置示例（Java 属性文件）如下：

# 聚合与缓冲
AggregationMaxCount = 100
AggregationMaxSize = 51200
RecordMaxBufferedTime = 100
RecordTtl = 30000

# 重试与限流
MaxConnections = 24
RequestTimeout = 6000
RecordMaxRetries = 5
RateLimit = 1000

# 指标与日志
MetricsLevel = DETAILED
LogLevel = INFO

其中 RecordMaxBufferedTime 决定单条记录在本地最多等待多久；RecordTtl 决定记录在缓冲区中的存活时间，超时未发送会被丢弃并计入指标；RateLimit 限制每秒发送的记录数，用于主动背压。实际调优时，如果下游分片写入受限，应优先降低 RateLimit 或增大 RecordMaxBufferedTime，而不是盲目增加重试次数。

使用 Kinesis Agent 采集日志

Kinesis Agent 通过配置文件定义数据源、目标流和转换规则。它支持文件轮转、断点续传和按行解析。一个典型的 agent 配置如下：

{
  "cloudwatch.emitMetrics": true,
  "kinesis.endpoint": "kinesis.cn-north-1.amazonaws.com.cn",
  "flows": [
    {
      "filePattern": "/var/log/app/*.log",
      "kinesisStream": "app-log-stream",
      "partitionKeyOption": "RANDOM",
      "dataProcessingOptions": [
        {
          "optionName": "LOGTOJSON",
          "logFormat": "COMMONAPACHELOG"
        }
      ]
    }
  ]
}

Agent 会自动处理文件偏移量，重启后从上次位置继续读取。需要注意的是，Agent 的 partitionKeyOption 若设为 RANDOM，记录会均匀分布到各分片；若设为 DETERMINISTIC，则按记录内容生成分区键，适合需要按业务键保序的场景。Agent 本身也具备重试机制，但遇到持续限流时，它会降低读取速率，形成天然背压。

使用 AWS SDK 精细控制写入

当业务需要精确控制分区键、重试策略和错误处理时，直接使用 AWS SDK 更合适。以下是一个 Python 示例，展示 PutRecords 的批量写入与失败重试：

import boto3, time, random
from botocore.exceptions import ClientError

client = boto3.client('kinesis', region_name='cn-north-1')

def put_batch(stream, records):
    resp = client.put_records(StreamName=stream, Records=records)
    failed = resp.get('FailedRecordCount', 0)
    if failed:
        retry = [r for i, r in enumerate(records)
                 if resp['Records'][i].get('ErrorCode')]
        return retry
    return []

def send_with_retry(stream, records, max_retries=5):
    attempt = 0
    pending = records
    while pending and attempt < max_retries:
        pending = put_batch(stream, pending)
        if pending:
            sleep = min(2 ** attempt + random.random(), 30)
            time.sleep(sleep)
            attempt += 1
    return pending  # 仍失败的记录

这段代码的关键点：put_records 返回的 FailedRecordCount 表示本次批量中失败条数；通过 ErrorCode 可以区分 ProvisionedThroughputExceededException（可重试）和 ValidationException（不可重试）。指数退避加上随机抖动，可以避免多个生产者同时重试造成二次冲击。

重试与背压的工程要点

重试不是越多越好。对于限流类错误，重试应配合退避；对于数据格式错误，重试不会成功，应直接进入死信队列或告警。建议设置最大重试次数（如 5 次）和最大重试时长（如 30 秒），超过后记录失败并上报指标。

背压的本质是生产速率超过分片写入能力。Kinesis 单分片写入上限为 1MB/s 或 1000 条记录/s。当 WriteProvisionedThroughputExceeded 指标持续上升时，说明生产端已经超过分片容量。此时可采取的措施包括：降低生产端并发或速率、增大聚合减少请求数、增加分片数量、切换到 On-Demand 模式，或在生产端本地缓冲并降速。

分区键与顺序性

分区键决定记录进入哪个分片。相同分区键的记录会进入同一分片并保持顺序。如果业务要求按用户 ID 保序，应使用用户 ID 作为分区键；如果只追求吞吐，可使用随机分区键。但要注意，热点分区键会导致单个分片过载，此时应通过加盐或复合键打散。

例如，将 userId 改为 userId + "#" + shardSalt，其中 shardSalt 取 0 到 N-1 的随机数，可以把同一用户的记录分散到多个分片，但会牺牲该用户的全局顺序。是否值得，取决于业务对顺序性的要求。

小结与检查清单

构建高可靠生产端时，可以按以下清单自查：是否根据场景选择了 KPL、Agent 或 SDK；是否设置了合理的聚合与缓冲参数；是否实现了指数退避重试并区分错误类型；是否监控了限流与失败指标；是否在背压时主动降速或扩容；是否根据顺序性要求选择了正确的分区键。把这些点落实到代码和配置中，生产端才能在流量波动和分片限流下保持稳定。