Glue流作业中Kinesis起始位置与检查点的组合作用疑问

问题描述 #

我正在使用Glue Streaming Job对接Kinesis Data Stream，希望作业在故障重启后始终从最后一条未处理记录开始读取。

当前代码如下：

df = glueContext.create_data_frame.from_option(
   connection_type="kinesis",
   connection_options={
       ... # 其他配置
       "startingPosition": "TRIM_HORIZON"
   }
)
glueContext.forEachBatch(
    frame=df,
    batch_function=batch_fn,
    options={
        "checkpointLocation": "s3://....",
        ...
    }
)

我知道TRIM_HORIZON会让作业从流的最开始读取，但Glue文档说明流作业用检查点而非作业书签跟踪已读取数据，所以想搞清楚这个配置组合的实际作用：是会从头读取但自动跳过已处理的数据？还是会重复处理历史数据？

回答 #

核心结论 #

当你同时配置startingPosition: "TRIM_HORIZON"和有效的checkpointLocation时，作业首次启动会从流的最开始读取，但故障重启后会直接从检查点记录的最后未处理位置继续读取，不会重复处理已完成的数据。

具体逻辑 #

• 首次启动：因为检查点目录是空的，Glue会遵循startingPosition的配置，从Kinesis流的起始位置（TRIM_HORIZON）开始消费数据，同时将处理进度持续写入指定的S3检查点目录。
• 故障重启：Glue会优先读取检查点目录中存储的消费偏移量，忽略startingPosition的配置，直接从上次中断时的最后未处理记录位置继续消费。

满足需求的优化建议 #

如果你希望无论首次启动还是重启，都尽量从最新的未处理位置开始（首次启动时不想从头读历史数据），可以把startingPosition改为"LATEST"，同时保留checkpointLocation配置：

df = glueContext.create_data_frame.from_option(
   connection_type="kinesis",
   connection_options={
       ... # 其他配置
       "startingPosition": "LATEST"
   }
)
glueContext.forEachBatch(
    frame=df,
    batch_function=batch_fn,
    options={
        "checkpointLocation": "s3://your-checkpoint-path",
        ...
    }
)

这样配置后：

• 首次启动：从Kinesis流的最新位置开始消费，不会读取历史数据
• 故障重启：从检查点记录的最后未处理位置继续，完全符合你的需求

内容的提问来源于stack exchange，提问作者Dark Matter