Spark Streaming Once Trigger模式下maxOffsetsPerTrigger未生效

针对你遇到的maxOffsetsPerTrigger设置为30M但实际拉取约60M记录的问题，结合Spark 2.4.8和Kafka连接器的特性，整理以下排查方向和解决方法：

可能原因 #

1. 参数解析错误：
   Spark 2.4.x的Kafka连接器中，maxOffsetsPerTrigger要求传入整数类型的偏移量数量，不支持M这类单位缩写。如果你的配置中传入的是字符串"30M"，Python API可能无法正确解析，导致实际生效的参数值超出预期。

2. Checkpoint残留数据干扰：
   如果之前运行过相同的流任务，checkpoint目录中可能残留旧的偏移量信息。使用trigger(once=True)时，Spark可能基于残留的偏移量范围，额外拉取超出maxOffsetsPerTrigger限制的数据。

3. Spark版本已知bug：
   Spark 2.4.8属于较旧版本，其Kafka连接器存在部分偏移量计算相关的bug，尤其是在trigger(once=True)模式下，可能导致maxOffsetsPerTrigger的限制未正确生效。

解决方法 #

• 确保参数为整数类型：
  将config.KAFKA_CONFIG["max_offsets_per_trigger"]设置为整数30000000（而非字符串"30M"），直接传递偏移量数量给Spark：

  .option("maxOffsetsPerTrigger", 30000000)
  

• 清理Checkpoint目录：
  删除config.GCS_STAGE_1_CHECKPOINT_LOCATION指向的GCS目录，确保流任务从全新状态开始运行，避免旧偏移量数据的干扰。

• 检查Spark日志确认参数生效情况：
  查看Spark Driver日志，搜索关键词maxOffsetsPerTrigger，确认实际加载的参数值是否为预期的30000000。如果日志显示参数值不符，需排查配置文件的加载逻辑。

• 升级Spark或Kafka连接器版本：
  考虑升级到Spark 2.4.x系列的最新补丁版本（或直接升级到Spark 3.x），新版本修复了多个Kafka连接器的偏移量计算bug，能更好地保证maxOffsetsPerTrigger参数的有效性。

• 手动控制分区偏移量（备选方案）：
  如果上述方法无效，可通过Kafka Admin API预先获取每个分区的起始和结束偏移量，手动计算每个分区可拉取的偏移量（总数量不超过30M），再通过startingOffsets和endingOffsets参数精准控制拉取范围，示例代码如下：

  import json
  # 假设已通过Kafka Admin API获取每个分区的起始偏移量start_offsets和结束偏移量end_offsets
  per_partition_limit = 30000000 // 20
  ending_offsets = {str(p): min(end_offsets[p], start_offsets[p] + per_partition_limit) for p in range(20)}
  
  streaming_df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", config.KAFKA_CONFIG["kafka_bootstrap_servers"]) \
  .option("subscribe", config.KAFKA_CONFIG["subscribe"]) \
  .option("failOnDataLoss", config.KAFKA_CONFIG["fail_on_data_loss"]) \
  .option("startingOffsets", json.dumps(start_offsets)) \
  .option("endingOffsets", json.dumps(ending_offsets)) \
  .load()
  

内容的提问来源于stack exchange，提问作者Yogesh Kumar