摘要Direct 模式上线前有两个问题绕不开并行度怎么调offset 怎么管。前者很多人栽在executor 加了不少吞吐还是上不去上——因为 Direct 模式的并行度天花板是 Kafka 分区数后者决定你能否做到 exactly-once。这篇把这两个问题一次讲清各给一套能落地的配置和代码。关键词Spark Streaming, Direct 模式, 并行度, Kafka 分区, offset 管理, exactly-once上篇并行度设置一、先记住一条铁律Direct 模式下RDD 的分区数 Kafka 的分区数每个 Kafka 分区对应一个 RDD 分区。这直接决定了并行度的上限。所以一个最常见的误区是executor 加了一堆、--executor-cores调得很大但 topic 只有 3 个分区——并行度还是 3一个 batch 最多就 3 个 task 在跑其余核全闲着。调并行度先看 Kafka 分区数别在 executor 数量上空转。二、三个决定并行的因子Kafka 分区数决定 RDD 分区数是并行度的天花板。executor 核数决定同时能跑多少 task。总核数要 ≥ 分区数否则 task 排队。maxRatePerPartition每个分区每个 batch 最多拉多少条用来限流/反压。三、调优目标和配置调优目标就一句一个 batch 的处理时间 batchInterval。做不到batch 就会堆积端到端延迟越来越大。处理跟不上时按顺序三招增加 Kafka 分区数提高并行度上限加 executor 核数让 task 不排队调大 batchInterval给每个 batch 更多时间。# 提交时设置 executor 核数spark-submit --executor-cores4--num-executors6...# 限流每分区每 batch 最多拉 10000 条# kafkaParams 里加spark.streaming.kafka.maxRatePerPartition-10000# 配合反压动态调整拉取速率spark.streaming.backpressure.enabled-true两个注意点核数要给后续 stage 留余地。拉数据的 task 只占一部分核shuffle、reduce 这些后续 stage 也要核。核数刚好等于分区数时后续 stage 会没核可用。Direct 模式没有blockInterval参数。那是 Receiver 模式切 block 用的Direct 模式调了没效果别搞混。下篇offset 管理四、三种方式选哪种方式一自动管理默认存 checkpoint什么都不用做Spark 自动把 offset 存进 checkpointJob 成功自动提交。前提是开了 checkpoint 且enable.auto.commitfalse。局限offset 和业务状态一起锁在 checkpoint 里没法独立回放、迁移。适合简单场景。方式二手动管理commitAsyncstream.foreachRDD{rddvaloffsetRangesrdd.asInstanceOf[HasOffsetRanges].offsetRanges// ... 处理 rdd写外部存储 ...stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)// 成功才提交}核心是处理成功才提交实现处理—提交的原子性这是 exactly-once 的实现方式。上一篇文章说过offsetRanges只能在第一个转换里取rdd.map(...)之后类型就丢了。方式三外部存储ZK / 数据库把 offset 存到 ZK 或数据库独立于 checkpoint。典型做法是把处理结果和 offset 放进同一个事务——结果写成功offset 才推进实现端到端 exactly-once。这是最严格、也是生产上对一致性要求高时的选择。五、指定 offset 回放有时候需要从某个位置重新消费比如补数据、事故回放用Assign策略 显式 fromOffsetvalfromOffsets:Map[TopicPartition,Long]Map(newTopicPartition(topic-a,0)-1000L,newTopicPartition(topic-a,1)-2000L)valstreamKafkaUtils.createDirectStream[String,String](ssc,PreferConsistent,ConsumerStrategies.Assign[String,String](fromOffsets.keys.toList,kafkaParams,fromOffsets))fromOffsets指定每个分区从哪个 offset 开始消费配合外部存储的 offset 就能做精确回放。六、总结并行度铁律RDD 分区数 Kafka 分区数调并行度先加 Kafka 分区别空加 executor。三个并行因子Kafka 分区上限、executor 核数不排队、maxRatePerPartition限流。调优目标一个 batch 处理时间 batchInterval核数要给后续 stage 留余地。offset 三种方式自动存 checkpoint简单、手动commitAsync处理成功才提交、外部存储结果offset 同事务端到端 exactly-once。需要回放时用 Assign fromOffsets 指定起始位置。作者大数据技术实践者博客blog.starzy.cnGitHubstarzy1990.github.io专注 AI Agent · LangGraph · RAG · 大数据架构 · 数据工程实践
