1. 为什么手动提交偏移量值得你花时间折腾把 Spark Streaming 和 Kafka 搭在一起做实时数据处理是很多数据管道项目的标配组合。但真正在生产环境跑起来之后你会发现一个让人头疼的问题数据要么重复消费要么莫名其妙丢了。这背后的核心原因往往就出在偏移量offset的管理方式上。Spark Streaming 消费 Kafka 数据时偏移量的提交方式直接决定了数据处理的语义——是至少一次还是至多一次能不能做到精确一次。默认情况下Spark Streaming 的 Direct API 并不会自动帮你把偏移量写回 Kafka它只是把偏移量存在内存或者 checkpoint 里。一旦任务重启如果 checkpoint 没配好或者被清掉了那就只能从头开始消费或者从某个不确定的位置继续数据一致性根本没法保证。手动提交偏移量的价值就在这里你能精确控制什么时候提交提交到哪里提交什么值。比如你想在数据真正写入下游存储之后再提交偏移量这样即使中间挂了重启后也能从上次成功处理的位置继续不会丢数据。再比如你想把偏移量存到外部数据库MySQL、Redis、ZooKeeper 都行方便监控和运维排查手动提交给了你完全的自由度。这篇文章适合谁看如果你已经在用 Spark Streaming 对接 Kafka但对偏移量管理还是一知半解或者你正准备搭建一条实时数据管道担心数据丢失和重复问题那这篇内容就是为你准备的。我会从核心原理讲起把每一步操作拆开揉碎再分享几个我在实际项目中踩过的坑和总结出来的经验。代码基于 Spark 2.x 和 Kafka 0.10 的 Direct API 来写这套组合目前仍然是很多存量系统的运行基础理解它对排查老系统的线上问题特别有帮助。2. Direct API 模式下偏移量到底存在哪2.1 Receiver 模式和 Direct 模式的本质区别要搞清楚手动提交偏移量这件事得先明白 Spark Streaming 消费 Kafka 有两种模式Receiver 模式和 Direct 模式。早期大家用的是 Receiver 模式它用一个 Receiver 线程去 Kafka 拉数据拉到的数据先缓存到 Executor 内存里然后 Spark Streaming 再通过 WALWrite Ahead Log来保证容错。这个模式的问题很明显Receiver 挂了数据可能丢WAL 又带来额外的写放大而且偏移量是自动提交到 ZooKeeper 的你根本控制不了提交时机。Direct 模式也叫 Direct Kafka API从 Spark 1.3 开始引入它彻底改变了玩法。Direct 模式下Spark Streaming 的每个 batch 直接去 Kafka 查询每个 partition 的最新偏移量范围然后生成一个 RDDRDD 的每个 partition 对应 Kafka 的一个 partition。数据不再经过 Receiver 中转而是由 Executor 直接消费。这样做的好处是第一不需要 WAL减少了写开销第二偏移量由你自己管理想什么时候提交就什么时候提交第三每个 Kafka partition 和 RDD partition 一一对应并行度天然对齐。关键点来了Direct 模式下Spark Streaming不会自动把偏移量提交回 Kafka。它默认把偏移量存在 checkpoint 目录里靠 checkpoint 来恢复。但 checkpoint 有几个坑一是 checkpoint 目录一旦损坏或者被清理偏移量就丢了二是 checkpoint 里存的是序列化的整个 DStream 图版本升级或者代码改动后可能反序列化失败三是你没法方便地查看当前消费到哪了。所以生产环境更推荐手动把偏移量提交到外部存储。2.2 偏移量提交的时机决定了数据语义偏移量提交的时机直接决定了你的数据管道是至少一次还是至多一次语义。我画个简单的逻辑帮你理解先处理数据再提交偏移量如果处理成功但提交失败重启后会重复消费这批数据这就是至少一次at-least-once。数据不会丢但可能重复。先提交偏移量再处理数据如果提交成功但处理失败重启后这批数据就跳过了这就是至多一次at-most-once。数据可能丢但不会重复。把偏移量和处理结果放在同一个事务里提交这就是精确一次exactly-once需要下游存储支持事务实现复杂度最高。大多数场景下我们追求的是至少一次因为数据重复可以通过下游去重解决但数据丢了就找不回来了。所以手动提交偏移量的标准做法是先完成数据处理逻辑确认写入下游成功后再提交偏移量。这个顺序千万不能反。2.3 偏移量存储位置的选择手动提交偏移量你得决定存哪。常见的选择有这么几种存储位置优点缺点适用场景Kafka 内部 topic无需额外组件Kafka 自带查询不方便需要额外消费简单场景不想引入外部依赖ZooKeeperKafka 老版本默认性能一般运维复杂老系统兼容MySQL查询方便可做监控需要额外维护数据库需要可视化监控的场景Redis读写快适合高频更新持久化需配置对性能要求高的场景HBase适合大规模偏移量运维成本高超大规模集群我在实际项目中用得最多的是 MySQL 和 Redis。MySQL 的好处是可以用 SQL 直接查某个 topic 的某个 partition 当前消费到哪了做监控大盘特别方便。Redis 的好处是写入快适合 batch 间隔很短比如 5 秒一个 batch的场景不会因为提交偏移量成为瓶颈。提示不管你选哪种存储一定要保证提交偏移量的操作是幂等的。也就是说同一个偏移量提交多次结果应该是一样的。这样即使重试也不会出问题。3. 从零搭建环境准备与依赖配置3.1 版本匹配这件事比你想的重要Spark 和 Kafka 的版本匹配是个老生常谈但又特别容易翻车的地方。我见过太多人因为版本不匹配跑起来各种 ClassNotFoundException 或者 NoSuchMethodError。这里给你一个我验证过的组合参考Spark 2.4.x Kafka 0.10.x 或 1.x 或 2.x用spark-streaming-kafka-0-10这个包Spark 3.x Kafka 2.x 或 3.x同样用spark-streaming-kafka-0-10但要注意 Scala 版本对应Maven 依赖大概长这样dependency groupIdorg.apache.spark/groupId artifactIdspark-streaming-kafka-0-10_2.11/artifactId version2.4.8/version /dependency dependency groupIdorg.apache.spark/groupId artifactIdspark-streaming_2.11/artifactId version2.4.8/version /dependency注意_2.11这个后缀它表示 Scala 版本。你的 Spark 集群用的哪个 Scala 版本这里就得对应上。Scala 2.11 和 2.12 的包是不兼容的搞错了直接报错。3.2 Kafka 参数配置的取舍创建 Kafka 消费流的时候有一堆参数要配。我把最关键的几个拎出来说说val kafkaParams Map[String, Object]( bootstrap.servers - kafka-broker1:9092,kafka-broker2:9092, key.deserializer - classOf[StringDeserializer], value.deserializer - classOf[StringDeserializer], group.id - my-consumer-group, auto.offset.reset - earliest, enable.auto.commit - (false: java.lang.Boolean) )这里有几个点需要特别注意enable.auto.commit必须设为 false。这是手动提交的前提。如果设成 trueKafka 客户端会自己在后台定期提交偏移量你的手动提交就乱套了。这个参数默认是 true很多人忘了改结果发现手动提交不生效排查半天。auto.offset.reset决定首次消费位置。可选值有earliest、latest、none。earliest表示从最早的消息开始消费latest表示从最新的开始。如果你手动管理偏移量这个参数只在找不到已提交偏移量时生效。比如你第一次启动任务或者偏移量存储被清空了它才会起作用。生产环境建议设成earliest避免漏数据。group.id要和你的偏移量存储对应。如果你把偏移量存在 MySQL 里那 group.id 就是你查询偏移量的 key 之一。同一个 group.id 在不同任务里要一致否则会重复消费。3.3 创建流的两种方式Direct API 创建流有两种方式区别在于你怎么指定起始偏移量方式一从最新位置开始val stream KafkaUtils.createDirectStream[String, String]( ssc, PreferConsistent, Subscribe[String, String](topics, kafkaParams) )这种方式下Spark 会去查你提交的偏移量如果enable.auto.commit为 false 且你没手动提交过那就看auto.offset.reset。PreferConsistent表示尽量把 partition 均匀分配到各个 Executor 上。方式二手动指定起始偏移量val fromOffsets Map( new TopicPartition(my-topic, 0) - 1000L, new TopicPartition(my-topic, 1) - 2000L ) val stream KafkaUtils.createDirectStream[String, String]( ssc, PreferConsistent, Assign[String, String](fromOffsets.keys.toList, kafkaParams, fromOffsets) )这种方式适合你从外部存储读取偏移量后精确指定从哪开始消费。比如你从 MySQL 里查到了上次消费的位置就用Assign来指定。注意Assign方式下Spark 不会去查 Kafka 的偏移量完全按你给的来。所以如果你给的偏移量超出了 Kafka 当前的范围可能会报错或者行为异常要做好边界检查。4. 手动提交偏移量的完整实现路径4.1 核心思路把偏移量和数据处理绑定手动提交偏移量的核心逻辑其实不复杂就三步从 DStream 里取出每个 batch 的偏移量范围处理数据写入下游下游写入成功后把偏移量提交到外部存储但这里有个关键问题Spark Streaming 的 DStream 操作是分布式的foreachRDD里的代码在 Driver 端执行而实际的数据处理在 Executor 端。偏移量的提交必须在 Driver 端做因为只有 Driver 知道这个 batch 处理完了没有。所以标准的代码结构是这样的stream.foreachRDD { rdd // 获取这个 batch 的偏移量范围 val offsetRanges rdd.asInstanceOf[HasOffsetRanges].offsetRanges // 触发数据处理Action 操作 val processed rdd.map(record { // 解析和处理数据 processRecord(record.value()) }).collect() // 或者写入下游 // 数据处理成功后提交偏移量 commitOffsets(offsetRanges) }这里有个坑rdd.map(...).collect()这种写法会把数据拉到 Driver 端如果数据量大Driver 直接 OOM。正确的做法是把数据写入下游存储比如 HBase、MySQL、Kafka写入操作本身是分布式的然后在 Driver 端等所有任务完成后再提交偏移量。4.2 用 foreachPartition 优化写入性能如果你要把数据写入外部存储用foreachPartition比foreach效率高得多。因为foreachPartition每个 partition 只建立一个连接而foreach每条记录都要建连接网络开销巨大。stream.foreachRDD { rdd val offsetRanges rdd.asInstanceOf[HasOffsetRanges].offsetRanges rdd.foreachPartition { partition // 每个 partition 建立一个连接 val connection createConnection() partition.foreach { record // 写入数据 writeToStorage(connection, record.value()) } connection.close() } // 所有 partition 处理完后提交偏移量 commitOffsets(offsetRanges) }但这里有个隐患foreachPartition是惰性执行的它不会阻塞 Driver。也就是说rdd.foreachPartition这行代码执行完Driver 不会等 Executor 把数据写完就继续往下走了直接去提交偏移量。如果这时候 Executor 写入失败了偏移量已经提交了数据就丢了。解决办法是用rdd.foreachPartition之后加一个 Action 操作来触发实际执行或者用rdd.mapPartitions(...).count()这种方式强制触发。更稳妥的做法是把写入逻辑放在foreachRDD里用rdd.foreachPartition写入然后通过累加器或者直接调用下游的同步接口来确认写入成功。4.3 偏移量提交到 MySQL 的完整代码我以 MySQL 为例给你一份完整的偏移量提交代码。先建表CREATE TABLE kafka_offsets ( topic VARCHAR(255) NOT NULL, partition INT NOT NULL, until_offset BIGINT NOT NULL, group_id VARCHAR(255) NOT NULL, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (topic, partition, group_id) );提交偏移量的函数def commitOffsets(offsetRanges: Array[OffsetRange], groupId: String): Unit { val connection DriverManager.getConnection( jdbc:mysql://localhost:3306/realtime, user, password ) val sql INSERT INTO kafka_offsets (topic, partition, until_offset, group_id) VALUES (?, ?, ?, ?) ON DUPLICATE KEY UPDATE until_offset VALUES(until_offset) val preparedStatement connection.prepareStatement(sql) offsetRanges.foreach { range preparedStatement.setString(1, range.topic) preparedStatement.setInt(2, range.partition) preparedStatement.setLong(3, range.untilOffset) preparedStatement.setString(4, groupId) preparedStatement.addBatch() } preparedStatement.executeBatch() preparedStatement.close() connection.close() }这里用了ON DUPLICATE KEY UPDATE保证同一个 topicpartitiongroup_id 的记录是更新而不是插入天然幂等。读取偏移量的函数def readOffsets(topics: Array[String], groupId: String): Map[TopicPartition, Long] { val connection DriverManager.getConnection( jdbc:mysql://localhost:3306/realtime, user, password ) val sql SELECT topic, partition, until_offset FROM kafka_offsets WHERE group_id ? val preparedStatement connection.prepareStatement(sql) preparedStatement.setString(1, groupId) val resultSet preparedStatement.executeQuery() val offsets scala.collection.mutable.Map[TopicPartition, Long]() while (resultSet.next()) { val topic resultSet.getString(topic) val partition resultSet.getInt(partition) val untilOffset resultSet.getLong(until_offset) offsets.put(new TopicPartition(topic, partition), untilOffset) } resultSet.close() preparedStatement.close() connection.close() offsets.toMap }启动流的时候先读偏移量如果有记录就用Assign方式指定起始位置val fromOffsets readOffsets(Array(my-topic), my-consumer-group) val stream if (fromOffsets.nonEmpty) { KafkaUtils.createDirectStream[String, String]( ssc, PreferConsistent, Assign[String, String](fromOffsets.keys.toList, kafkaParams, fromOffsets) ) } else { KafkaUtils.createDirectStream[String, String]( ssc, PreferConsistent, Subscribe[String, String](Array(my-topic), kafkaParams) ) }4.4 提交到 Kafka 内部 topic 的方式如果你不想引入外部存储也可以把偏移量提交回 Kafka 自己的__consumer_offsetstopic。这种方式下你不需要自己写存储逻辑直接调用 Kafka 的 API 就行stream.foreachRDD { rdd val offsetRanges rdd.asInstanceOf[HasOffsetRanges].offsetRanges // 处理数据 rdd.foreachPartition { partition // 写入下游 } // 提交偏移量到 Kafka stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges) }commitAsync是异步提交不会阻塞。但异步提交有个问题如果提交失败了你不会立刻知道。所以生产环境建议用同步提交或者至少加个回调处理失败情况。提示提交到 Kafka 内部 topic 的方式偏移量查询不太方便。你需要用 Kafka 自带的工具或者写个消费者去读__consumer_offsets运维排查时比较麻烦。如果团队没有专门的 Kafka 运维工具还是建议存到 MySQL 或 Redis。5. 那些让我熬夜排查的坑5.1 序列化问题Task not serializable这个错误我敢说每个写 Spark 的人都遇到过。当你把偏移量提交逻辑写在foreachRDD里而提交逻辑又引用了外部对象比如数据库连接池、配置对象Spark 会尝试把这些对象序列化后发到 Executor。如果这些对象不支持序列化就报Task not serializable。解决办法有几个一是把不需要序列化的对象声明为transient二是把提交逻辑抽成一个独立的单例对象或者静态方法三是用foreachPartition在 Executor 端创建连接而不是在 Driver 端创建好再传过去。我个人的习惯是所有跟数据库连接相关的操作都在foreachPartition内部创建连接Driver 端只负责提交偏移量这种轻量操作。这样既避免了序列化问题也避免了连接在网络上传输。5.2 偏移量提交了但数据没写完这是最危险的情况直接导致数据丢失。根本原因就是前面说的foreachPartition是惰性的Driver 不等 Executor 写完就提交了偏移量。我踩过一次这个坑任务跑得好好的突然有一天发现下游数据少了一批。排查了半天发现是某个 Executor 在写入时抛了异常但 Driver 已经把偏移量提交了重启后直接从下一批开始消费那批数据就永远丢了。修复方案是确保数据处理和偏移量提交在同一个事务边界内。具体做法有两种方案一用累加器确认写入成功val successCounter ssc.sparkContext.longAccumulator(success) val failCounter ssc.sparkContext.longAccumulator(fail) stream.foreachRDD { rdd val offsetRanges rdd.asInstanceOf[HasOffsetRanges].offsetRanges rdd.foreachPartition { partition partition.foreach { record try { writeToStorage(record.value()) successCounter.add(1) } catch { case e: Exception failCounter.add(1) // 记录失败详情 } } } // 只有全部成功才提交偏移量 if (failCounter.value 0) { commitOffsets(offsetRanges) } else { // 有失败不提交让这批数据重新处理 throw new RuntimeException(数据处理失败不提交偏移量) } }方案二把偏移量和数据一起写入下游如果下游是支持事务的数据库比如 MySQL你可以把偏移量和处理结果放在同一个事务里提交。这样要么都成功要么都失败天然保证一致性。这种方式实现精确一次语义但要求下游存储支持事务。5.3 checkpoint 和手动提交的冲突如果你同时开了 checkpoint 和手动提交偏移量可能会遇到一些奇怪的行为。Spark Streaming 的 checkpoint 会保存 DStream 的元数据和偏移量信息。当你从 checkpoint 恢复时它会用 checkpoint 里的偏移量而不是你手动提交的偏移量。这就导致一个问题你手动提交的偏移量可能比 checkpoint 里的更新但恢复时却用了旧的。解决办法是如果用了手动提交就不要再依赖 checkpoint 来恢复偏移量。你可以关掉 checkpoint或者用ssc.getState()来判断是否从 checkpoint 恢复然后手动覆盖偏移量。我的建议是生产环境如果用了手动提交偏移量就干脆别开 checkpoint。checkpoint 主要用于 Driver 故障恢复但如果你把偏移量存在外部Driver 重启后从外部读偏移量就行不需要 checkpoint。这样还避免了 checkpoint 目录膨胀和反序列化失败的问题。5.4 分区再均衡导致的偏移量错乱Kafka 的消费者组会发生分区再均衡rebalance比如你增加或减少了消费者实例或者某个消费者挂了。再均衡期间分区会重新分配偏移量的提交可能会错乱。在 Spark Streaming 的 Direct 模式下每个 batch 的偏移量范围是在 Driver 端确定的。如果再均衡发生在 batch 处理期间可能会导致某些 partition 的数据被重复消费或者跳过。缓解办法是设置合理的session.timeout.ms和heartbeat.interval.ms减少不必要的再均衡。另外在提交偏移量之前检查一下当前分配的 partition 是否和 batch 开始时一致如果不一致就跳过提交等下一个 batch 再处理。6. 监控与运维让偏移量看得见6.1 用 MySQL 做偏移量监控大盘把偏移量存在 MySQL 的一个好处是你可以很方便地做监控。比如查每个 partition 的消费延迟SELECT o.topic, o.partition, o.until_offset, (SELECT MAX(offset) FROM kafka_latest_offsets l WHERE l.topic o.topic AND l.partition o.partition) - o.until_offset AS lag FROM kafka_offsets o WHERE o.group_id my-consumer-group;这个 lag 就是消费延迟。如果 lag 持续增长说明消费速度跟不上生产速度需要考虑扩容或者优化处理逻辑。你还可以用 Grafana 或者自己写个简单的 Web 页面定时查这个 SQL把 lag 画成曲线图。这样一眼就能看出任务是否健康。6.2 偏移量提交失败的告警偏移量提交失败是个严重问题必须告警。常见的失败原因有数据库连接断了、网络抖动、表被锁了。你可以在提交偏移量的代码里加 try-catch失败时发告警邮件、短信、钉钉都行。try { commitOffsets(offsetRanges) } catch { case e: Exception // 发送告警 sendAlert(s偏移量提交失败: ${e.getMessage}) throw e // 重新抛出让 Spark 重试 }注意这里要重新抛出异常让 Spark 知道这个 batch 失败了它会根据配置重试。如果你吞掉了异常Spark 以为成功了就不会重试数据就丢了。6.3 定期清理过期的偏移量记录偏移量表如果一直不清理会越来越大。虽然每个 topicpartition 只有一条记录因为有ON DUPLICATE KEY UPDATE但如果你的 topic 很多或者经常换 group.id表还是会膨胀。建议定期清理不再使用的 group.id 的记录。比如你可以加一个updated_at字段定期删除超过 30 天没更新的记录。或者干脆在换 group.id 的时候手动清理旧记录。7. 几个能直接抄的配置模板7.1 完整的 Spark Streaming 任务模板object KafkaOffsetDemo { def main(args: Array[String]): Unit { val conf new SparkConf() .setAppName(KafkaOffsetDemo) .set(spark.streaming.kafka.maxRatePerPartition, 1000) .set(spark.streaming.backpressure.enabled, true) val ssc new StreamingContext(conf, Seconds(10)) val kafkaParams Map[String, Object]( bootstrap.servers - kafka-broker:9092, key.deserializer - classOf[StringDeserializer], value.deserializer - classOf[StringDeserializer], group.id - demo-group, auto.offset.reset - earliest, enable.auto.commit - (false: java.lang.Boolean), max.poll.records - 500 ) val topics Array(demo-topic) val fromOffsets OffsetStore.readOffsets(topics, demo-group) val stream if (fromOffsets.nonEmpty) { KafkaUtils.createDirectStream[String, String]( ssc, PreferConsistent, Assign[String, String](fromOffsets.keys.toList, kafkaParams, fromOffsets) ) } else { KafkaUtils.createDirectStream[String, String]( ssc, PreferConsistent, Subscribe[String, String](topics, kafkaParams) ) } stream.foreachRDD { rdd val offsetRanges rdd.asInstanceOf[HasOffsetRanges].offsetRanges // 处理数据 val processed rdd.map(record processRecord(record.value())) // 写入下游这里用 count 触发执行实际替换为写入逻辑 val count processed.count() // 提交偏移量 if (count 0) { OffsetStore.commitOffsets(offsetRanges, demo-group) } } ssc.start() ssc.awaitTermination() } def processRecord(value: String): String { // 实际处理逻辑 value.toUpperCase } }7.2 关键参数速查表参数推荐值说明spark.streaming.kafka.maxRatePerPartition1000-5000每个 partition 每秒最大消费条数防止第一批数据过多spark.streaming.backpressure.enabledtrue开启背压根据处理能力动态调整消费速率spark.streaming.kafka.consumer.poll.ms512消费者 poll 超时时间max.poll.records500每次 poll 最大记录数太大容易导致处理超时session.timeout.ms30000消费者会话超时再均衡相关heartbeat.interval.ms10000心跳间隔一般是 session.timeout 的 1/37.3 背压机制的实际效果背压backpressure是 Spark Streaming 1.5 引入的特性它可以根据上一个 batch 的处理速度动态调整下一个 batch 的消费速率。这个机制在数据流量波动大的场景下特别有用。我实测下来开启背压后任务在流量高峰时不会因为一批数据太多而 OOM而是会自动降低消费速率等处理完了再加速。但背压也不是万能的如果下游存储本身写入很慢背压只能缓解不能根治。这时候还是得优化下游写入逻辑或者增加并行度。注意背压和maxRatePerPartition可以同时使用。maxRatePerPartition设的是上限背压在这个上限内动态调整。如果两个都不设第一批数据可能会非常大直接打爆内存。8. 关于偏移量管理的一些个人体会偏移量管理这件事说简单也简单说复杂也复杂。简单在于核心逻辑就那几步读偏移量、处理数据、提交偏移量。复杂在于各种边界情况和异常处理稍不注意就丢数据或者重复消费。我个人的经验是宁可重复不可丢失。重复消费可以通过下游去重解决比如用消息里的唯一 ID 做幂等写入。但数据丢了就真找不回来了。所以提交偏移量的时机一定要放在数据处理成功之后哪怕这样会导致一些重复。另外偏移量的存储选择要根据团队实际情况来。如果团队有成熟的 MySQL 运维能力那就存 MySQL查询方便。如果追求极致性能那就存 Redis。如果不想引入外部依赖那就提交回 Kafka。没有绝对的好坏只有适不适合。最后说一个容易被忽略的点偏移量的监控和告警。很多人把偏移量提交了就完事了从来不关注提交是否成功、消费是否有延迟。等到发现数据不对的时候可能已经丢了好几天的数据了。所以一定要做监控把 lag 和提交成功率都监控起来有问题第一时间知道。这套方案我在几个项目里都用过跑得挺稳。当然每个项目的场景不一样你可能需要根据自己的情况调整。比如 batch 间隔、并行度、存储选型这些都得结合实际数据量和集群规模来定。但核心思路是不变的先处理数据再提交偏移量保证至少一次语义然后通过监控和告警兜底。