Spark Streaming对接Kinesis故障重启及限流问题咨询

1. 单批次按字节维度限制处理数据量的可行方案 #

共有三层可落地的配置/实现方案，优先使用原生配置，兜底用应用层逻辑：

• 拉取侧硬限流：构造Kinesis InputDStream时，通过builder传入fetchSizeBytes参数，直接控制单次从单个Kinesis分片拉取的最大字节数。可根据集群单批次最大承载字节数，结合运行的Receiver总数、15s批次间隔内单Receiver的拉取次数，倒推配置该值，从拉取源头控制单批次流入的数据量。
• 动态反压限流：开启Spark Streaming原生反压机制，配置spark.streaming.backpressure.enabled=true，同时配置EMR Kinesis连接器专属参数spark.streaming.kinesis.rateLimitBytePerSecond，设置单分片每秒最大可消费的字节阈值。该机制会根据历史批次的处理延迟、调度延迟动态调整实际拉取速率，避免固定阈值在流量波动时出现限流过松/过紧的问题。注意不要只用默认的maxRatePerPartition参数，该参数是按消息条数限流，无法匹配按字节控制的需求。
• 应用层兜底限流：可在DStream处理链路的最前端添加transform算子，实时累计当前批次所有数据的总字节数，达到预设安全阈值时，将超出部分的数据缓存后合并到下一批次处理，避免极端场景下原生配置失效导致的雪崩。

2. 宕机重启后积压存量数据+实时新增数据的调度逻辑 #

首先明确配置生效规则：InitialPositionInStream=TRIM_HORIZON仅在应用首次启动、DynamoDB中不存在对应消费组的checkpoint位点记录时生效。只要应用之前成功运行过、DynamoDB中留存了checkpoint记录，重启时会直接从最后一次成功提交的位点开始消费，和TRIM_HORIZON配置无关，不会从流的最早位置重新消费。
具体调度规则如下：

• 消费顺序：严格按照Kinesis流内数据的写入时间顺序消费，优先拉取处理宕机期间产生的存量积压数据，待存量积压完全追平后，才会进入实时新增数据的消费状态，不会出现存量数据和实时数据穿插处理的情况。
• 批次切分规则：重启后会按照配置的15s批次间隔，把从最后checkpoint位点到当前最新位点之间的所有存量数据，切分为多个固定15s窗口的批次，按时间顺序依次进入Spark调度队列排队。以宕机2小时为例，总共会切分出480个待处理存量批次，集群处理完前一个批次后，才会拉取调度下一个批次。
• 积压追平逻辑：如果集群正常消费实时数据时，单批次处理耗时小于15s的批次间隔，追积压时每处理完一个15s的存量批次，就能追平对应时长差的积压进度。比如单批次处理耗时5s，每轮处理就能追上10s的积压，持续运行直到存量数据全部处理完成，回到正常消费实时数据的状态。
• 注意事项：
  • 未开启按字节限流和反压时，每个存量批次会尝试拉取对应15s窗口内的全量数据，如果积压窗口内的数据量超过集群单批次承载上限，就会触发重启后再次故障的雪崩问题。
  • 由于配置的Kinesis CheckPointInterval为60s，故障重启后最多会重复消费故障发生前最后60s内已经处理但未成功提交位点的数据，业务逻辑需要做好幂等处理。

内容的提问来源于stack exchange，提问作者Rishabh Raj