使用Spark(2.3)自定义Hive Streaming Sink时遇Offset乱序提交错误

首先，这个错误的根源是Spark内置的TextSocketSource有严格的offset递增校验逻辑：它要求每次提交的offset必须比上一次的大，否则就会抛出Offsets committed out of order异常。这种情况通常发生在任务重启后，SocketSource从头开始读取数据，但checkpoint里保存了之前已经提交过的更大offset，导致提交时冲突。

下面是针对不同场景的解决方案：

1. 测试场景：清理Checkpoint目录重启任务 #

因为TextSocketSource本身是用于测试的数据源，不具备容错能力——任务重启后它会从连接的socket开头重新读取数据，而checkpoint里还保留着上次任务运行时提交的offset（比如1），新任务提交offset 0时就触发了校验失败。

解决步骤：

• 先停止当前的流任务
• 删除配置中指定的checkpoint目录：rm -rf /checkpoints/employee/checkpoint
• 重新启动流任务

⚠️ 注意：这个操作会让任务重新消费所有socket发送的数据，仅适合测试环境，生产环境不建议使用SocketSource。

2. 检查自定义Hive Sink的Offset处理逻辑 #

虽然错误来自SocketSource，但也有可能是自定义的hive-streaming sink在offset管理上出现了异常，导致Spark尝试重复提交旧的offset。可以排查：

• 确认sink的commit方法是否正确实现了Spark Structured Streaming的Sink接口规范，是否正确将offset写入checkpoint
• 检查sink有没有出现异常回滚offset的情况，导致MicroBatchExecution认为需要提交之前的offset

3. 生产环境：替换为容错性数据源 #

TextSocketSource不支持容错，生产环境强烈建议使用像Kafka这样支持exactly-once语义的数据源。Kafka会保存消费的offset，任务重启后可以从checkpoint中恢复的offset位置继续消费，不会出现offset乱序的问题。

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