做实时计算这几年我越来越觉得 Kafka Spark Streaming 这套组合就像工地上最常见的“脚手架”——它不一定是最精致的方案但胜在到处都能搭、材料好找、坏了随时有人会修。去年负责某个订单实时分析任务时我们又在这套技术栈上踩了一圈监控大屏里处理延迟从十几秒一路爬到几分钟排查到最后发现不是框架不行而是我们对分区策略、偏移量提交和背压机制的理解还停留在“能跑”的层面。这篇文章就把这次实战里选型、写代码、调参、排障的全过程整理出来给正在用或准备用这套技术栈的同行一个参照。本文适合这几类人看刚接手 Spark Streaming 存量任务、想弄明白 Kafka 消费端到底该怎么写的人以及已经在跑实时任务、却被延迟告警和重复消费折磨过的同学。我会尽量把“为什么这样选”“为什么这里会出问题”讲透而不是只丢一堆配置项。另外多说一句如果你是新项目选型我会建议认真评估 Structured Streaming但如果你的线上任务已经是 Spark Streaming 且短期不打算迁移这篇文章里的经验仍然能直接派上用场。1. 实时计算选型为什么在这个场景我坚持用 KafkaSpark Streaming1.1 业务场景和需求边界先说背景。当时的业务方给了一个很明确的需求每天约八百万条订单数据高峰时段每分钟大约十万条写入需要实时统计销售额、按省份聚合订单量、并对异常订单做告警。延迟要求是“分钟级别可以接受”但数据不能丢、不能算错。这类需求放到今天候选方案无非是 Flink、Spark Streaming、Structured Streaming以及已经比较少人提起的 Storm。Storm 我是用过的延迟确实低但 API 偏底层窗口统计、状态管理、故障恢复这些都要自己拼装开发和运维成本都不低。Flink 的窗口和状态管理确实做得更好但当时团队里对 Flink 有实操经验的人屈指可数如果硬切光是把大家教会、把监控体系重建一遍就够折腾两三个月。相比之下Spark Streaming 的微批模型在这种场景下有明显优势五秒一个批次延迟完全在业务接受范围内DStream 上的 map、reduceByKeyAndWindow 这些算子和 Spark 批处理风格一致团队里写惯了 Spark SQL 的人几乎没有学习成本而且它可以复用已有的 Spark 集群调度、日志、监控体系都是现成的。Kafka 作为数据管道天然解决了削峰填谷的问题生产端把订单数据打到 Kafka消费端由 Spark Streaming 按自己的节奏拉取上下游互不拖累。最终我们确定的架构就是Kafka 负责缓冲和分发Spark Streaming 负责微批计算下游接 MySQL 和 Redis。1.2 为什么这套组合到今天仍不算过时很多人一听到 Spark Streaming 就摇头觉得它已经被 Flink“拍在沙滩上”了。从技术演进的角度看Structured Streaming 确实在 API 和语义上更先进Spark 官方也把 DStream 列为维护模式。但现实是大量存量任务仍然跑在 Spark Streaming 上原因很简单迁移成本。我见过不少团队线上用 Spark Streaming 跑着几十上百个实时任务数据口径、告警规则、下游表结构全是按这套框架设计的不是说换就能换的。而且从能力边界看Spark Streaming 在秒级延迟、高吞吐场景下完全够用。它不适合的是毫秒级低延迟场景但那种场景本来就应该选专用流处理引擎或自研方案而不是指望一套通用计算框架解决所有问题。关键还是想清楚自己的需求边界如果业务方要的是“五秒内看到结果”而不是“五百毫秒内响应”Kafka Spark Streaming 无论从稳定性还是运维成本上说都有足够的理由继续存在。2. 环境搭建与版本选型几个会让项目当场崩溃的细节2.1 版本矩阵别以为随便配就能跑Kafka 对接 Spark Streaming相关组件版本必须成对匹配否则跑起来各种奇怪报错。这里直接给出我验证过的组合Spark 2.4.x建议 2.4.8 Kafka client 2.x Scala 2.11 或 2.12。spark-streaming-kafka-0-10_2.11 这个 artifact 对应 Scala 2.11 版本如果你的 Spark 编译用的是 Scala 2.12那就把 artifactId 里的 _2.11 换成 _2.12否则运行时会报 Scala 版本冲突。我见过最典型的错误是直接引入 kafka-clients 依赖并指定一个比较新的版本结果和 Spark Streaming 内部的 Kafka 客户端实现冲突消费者一直报NoSuchMethodError。正确做法是只依赖 spark-streaming-kafka-0-10让它自己传递依赖不要手动覆盖 kafka-clients 版本除非你非常确定两个版本之间的 API 没有变化。另外Spark 2.4.x 有对应的 spark-streaming-kafka-0-10 版本是 2.4.8千万别拿 2.3.x 的包配 2.4.x 的 Spark运行时会报类不存在。dependency groupIdorg.apache.spark/groupId artifactIdspark-streaming-kafka-0-10_2.11/artifactId version2.4.8/version /dependency2.2 Kafka 集群参数和生产配置Kafka 侧主要关注两个参数分区数和副本因子。分区数的估算逻辑我后面会单独讲这里只说一个基本公式分区数至少等于目标吞吐除以单分区能支撑的消费吞吐。比如你的目标高峰吞吐是每分钟十万条单分区实测能扛住每分钟两万条那分区数就不能少于五。但实际配置时我建议留两倍以上余量因为分区数在创建 Topic 后虽然可以增加但会增加运维成本和跨分区数据倾斜的风险。副本因子在生产环境建议至少设成 2允许一个 broker 宕机不丢数据但如果你的机器资源紧张2 就够了不用盲目的三副本配置因为每多一个副本就意味着多一份磁盘和网络开销。我在测试环境见过有人为了稳妥把 replication-factor 设成 3结果三台 broker 每台磁盘都差点被打爆后来改成 2 才平衡下来。kafka-topics.sh --create \ --topic order_topic \ --partitions 12 \ --replication-factor 2 \ --bootstrap-server kf01:9092,kf02:9092,kf03:90923. 核心代码骨架从 Kafka 消费到窗口统计的完整写法3.1 KafkaUtils.createDirectStream 为什么是首选Spark Streaming 对接 Kafka 有两种方式基于 Receiver 的旧方式以及基于 Direct API 的新方式。新项目必须选 Direct API也就是KafkaUtils.createDirectStream。原因不复杂Receiver 方式先由 Executor 上的 Receiver 持续不断地拉取数据再把数据作为 Block 交给 Spark Streaming中间多了一层内存缓冲一旦 Receiver 挂了或者 Driver 重启就可能丢数据即使开启了 WAL 也还是会有重复计算的窗口。Direct API 则让每个批次直接去 Kafka 拉取对应分区的数据没有中间缓冲层RDD 分区和 Kafka 分区一一对应数据的可靠性完全由 Kafka 和偏移量管理来保证。另一个容易忽视的点是 LocationStrategies。它控制 Spark 从哪些 Executor 上去拉 Kafka 分区数据。默认的PreferConsistent会把分区尽可能均匀地分发到所有 Executor 上适合大多数场景。如果 Executor 和 Kafka broker 部署在同一批机器上可以考虑PreferBrokers能让数据在本地消费减少网络传输但这种优化收益通常只在集群规模很大时才明显。PreferFixed 则适合某些分区数据量极不均匀的场景可以手动指定分区到 Executor 的映射一般用不到。3.2 核心代码从消费到窗口统计下面是我在项目里使用的一段核心骨架代码已经简化掉业务字段。先初始化 StreamingContext设置批次间隔为五秒然后创建 DStreamval conf new SparkConf() .setAppName(order-realtime-analysis) .setIfMissing(spark.master, yarn) val ssc new StreamingContext(conf, Seconds(5)) val kafkaParams Map[String, Object]( bootstrap.servers - kf01:9092,kf02:9092,kf03:9092, key.deserializer - classOf[StringDeserializer], value.deserializer - classOf[StringDeserializer], group.id - order-realtime-job, auto.offset.reset - latest, enable.auto.commit - false ) val topics Array(order_topic) val stream KafkaUtils.createDirectStream[String, String]( ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) )拿到 DStream 之后解析每条订单数据。我们的消息格式是竖线分隔订单ID、省份ID、商品ID、金额、时间戳。你只需要按自己的消息格式调整这段解析逻辑val orders stream.map { record val fields record.value().split(\\|) // fields(0)订单ID fields(1)省份ID fields(2)商品ID fields(3)金额 (fields(1), fields(3).toDouble) }实时统计每个省份的销售额窗口长度 60 秒、滑动间隔 5 秒也就是每五秒统计一次最近一分钟的数据val provinceSales orders .reduceByKeyAndWindow(_ _, Seconds(60), Seconds(5)) provinceSales.foreachRDD { rdd rdd.foreachPartition { partition // 在这一步把结果写入外部存储 // 注意每个分区建立一个连接不要每条数据都 new 一个客户端 } }这段代码看起来简单但有几个地方我强烈建议你注意。第一次写实时任务的人最容易犯的错是在foreachRDD外面创建连接客户端然后在foreachPartition里直接用。这个“连接”会被序列化后分发到每个 Executor 上大概率出现两种结果要么连接对象根本不能被序列化直接报异常要么因为多线程共享连接导致数据错乱。正确做法是在foreachPartition内部创建连接用完即关配合连接池减少重复创建的开销。3.3 偏移量提交时机不对要么丢数要么重复偏移量提交是整个消费逻辑里最容易被低估的一环。我们把enable.auto.commit设成false就是为了自己控制提交时机。Direct API 获取当前批次的偏移量范围很简单但必须在 Driver 端拿不能在 Executor 端拿。下面这段代码是处理完成后手动提交的典型姿势stream.foreachRDD { rdd // 1. 在 Driver 端取出当前批次读取的偏移量范围 val offsetRanges rdd.asInstanceOf[HasOffsetRanges].offsetRanges // 2. 执行业务逻辑算指标、打告警、写外部存储 // ... 业务处理 ... // 3. 处理成功后再提交偏移量 stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges) }为什么“先处理、后提交”这么重要因为一旦你先提交了偏移量然后处理逻辑崩溃下次重启时就会直接从已提交的位置开始消费中间那段数据就丢了。反过来先处理再提交如果处理完还没来得及提交就挂了重启后会再从旧偏移量开始消费这段数据会被重复处理。所以这个选择其实就是“至少一次”和“至多一次”之间的取舍。我们选了先处理再提交配合下游的幂等写入来吸收重复数据这样既能保证不丢数又能把重复的影响控制住。具体怎么做幂等我会在下一节展开。4. 背压、偏移量与 Exactly Once三个必须直面的大山4.1 背压机制如何让消费速度匹配处理速度背压Backpressure是 Spark Streaming 里一个非常核心的机制。我第一次接手实时任务时把它关掉了结果高峰期一个批次的数据量暴增处理不过来后面的批次全部排队整个 Job 的延迟像滚雪球一样越滚越大。开启背压后Spark Streaming 会根据上一批任务的实际处理耗时来动态调整下一批的消费速率相当于给消费端装了一个“限流阀”。相关配置有三个我列一下配置项作用spark.streaming.backpressure.enabled是否开启背压建议 truespark.streaming.backpressure.initialRate启动时的初始消费速率条/秒避免启动瞬间打爆集群spark.streaming.kafka.maxRatePerPartition每个分区每秒最大消费条数是硬上限这段话翻译成实践建议先开背压再设初始速率再设一个合理的单分区上限。比如你的单分区处理能力实测是每秒 1500 条maxRatePerPartition 就设 1500 到 2000这样即使 Kafka 里积压了大量数据每个批次也不会一次吞太多处理延迟会更稳定。initialRate 设一个偏低的值比如 500让任务平稳启动后再慢慢爬坡。4.2 三种消费语义的代价权衡先明确一个概念在 Kafka Spark Streaming 这套架构下讨论 Exactly Once 不能只看消费端要分三个层面看——Kafka 消费、流计算、下游写入。At Most Once先提交偏移量再做业务处理宕机时丢数据适合对数据完整性要求极低的场景。At Least Once先处理再提交宕机时重复消费配合下游幂等可以做到“实际不重复”。Exactly Once从消费到计算到写入全程不丢不重。Spark Streaming 本身能从 Kafka 精确读取数据但这只能保证“消费端正好一次”计算端和写入端之间的重复仍要外部配合。所以真正可靠的做法是消费端采用 At Least Once下游写入采用幂等操作最终对外表现为 Exactly Once。这是一个工程方案不是某个参数能解决的。4.3 幂等输出把 Exactly Once 落地的现实方案幂等写入的思路是让同样的数据被写入多次时最终结果和写入一次完全一致。举几个我在项目中用过的例子MySQL 写入表里建唯一键比如订单ID。写入使用INSERT ... ON DUPLICATE KEY UPDATE重复数据只会更新同一行不会产生双份。HBase 写入以订单ID或业务ID作为 rowkey天然就是幂等的同一条数据写多少次都只有一个版本。Redis 写入用SET NX或SET覆盖看你的指标口径是累加还是直接赋值。用一句话总结就是不要试图让流管道自身做到完美无缺的 exactly-once那样成本极高且难以调试。把力气花在下游去重上简单、稳定、效果好。这也是我对每个做实时计算项目的人的建议先把“重复消费会不会造成数据错误”这个问题想清楚再去研究各种高级语义。5. 线上调优分区、批次与 GC 的一组实测5.1 分区数、并行度与 repartition 的取舍Kafka 的每个分区对应 Spark RDD 的一个分区。所以消费端的并行度上限天然等于 Kafka Topic 的分区数。如果一台 Executor 有三个核而你有十二个分区和两个 Executor每个 Executor 会同时消费三个分区每个分区分配到一个核。这时候瓶颈往往不是 CPU而是单分区的消费速率——因为每个核在同一个时刻只能处理一个分区。常见误区是发现处理不过来就盲目对 DStream 做 repartition把数据分到更多分区。这样做的确能增加下游并行度比如在reduceByKeyAndWindow之前把 12 个分区 repartition 成 36 个后面的聚合能跑得更快。但代价是引入了一次 shuffle数据要跨 Executor 传输增加网络和磁盘 IO如果数据量本身就大shuffle 的开销反而会拖慢整个批次。我建议的做法是先保证 Kafka 分区数和 Executor 的并行度匹配比如两个 Executor、每个 Executor 三个核那分区数最好在 12 到 24 之间。实在要 repartition也只在单个算子处理明显倾斜时用不要全局无脑加。5.2 批次间隔与状态内存的搭配batchDuration 是最重要的微批参数。它决定了每个批次的数据量、处理延迟下限以及窗口计算需要缓存多少数据。设置时请记住一个经验法则批次处理时间应当小于批次间隔的一半。比如 5 秒的批次处理时间最好在 2.5 秒以内如果经常跑到 4 秒以上说明资源已经吃紧迟早会积压。窗口计算对内存的消耗比较容易估算。以我们的场景为例batchDuration 5 秒窗口长度 60 秒那一个窗口需要保留 12 个批次的中间结果。当你同时跑多个窗口、多个 key 的聚合时内存占用会成倍增加。所以一旦开了窗口计算Executor 的内存就要预留出足够余量否则频繁 Full GC 会一瞬间打垮整条链路。5.3 G1GC 与一组调优实测数据Spark Streaming 任务对 GC 的敏感度比普通批处理高得多。因为微批模型要求每 5 秒就要完成一次完整的调度、计算和提交任何一次长时间 GC 都可能让该批次超时。我用 G1GC 替换默认的 Parallel GC 后高峰期平均处理时间明显回落。提交 Spark 任务时加上这些参数spark-submit \ --conf spark.executor.extraJavaOptions-XX:UseG1GC -XX:G1HeapRegionSize16m \ --conf spark.streaming.backpressure.enabledtrue \ --conf spark.streaming.backpressure.initialRate800 \ --conf spark.streaming.kafka.maxRatePerPartition1500 \ --class com.example.OrderRealtimeAnalysis \ order-realtime-analysis.jar下面是一组我们在测试环境和生产环境对比过的数据场景是 5 秒批次、约 60 个 Kafka 分区、高峰每分钟约十万条订单参数调优前调优后Executor 数816单 Executor 核数44背压关闭开启maxRatePerPartition未设置1500G1GC否是高峰期平均处理时间5.8 秒/批2.4 秒/批高峰期最大延迟持续累积稳定在 30 秒内调优永远不是单点改动而是组合拳。背压打开了批次处理时间降下来GC 换了长暂停消失分区并行度也上来了。三者叠加才把延迟从“持续累积”变成“稳定可控”。6. 一次延迟告警的完整排查链路6.1 现象延迟从 20 秒涨到 180 秒有一天晚上十点左右值班群收到告警实时任务的消费延迟持续攀升Kafka 消费者组里的 Lag 越来越大。打开监控大屏处理延迟已经从正常的 20 秒左右涨到 180 秒而且还在往上走。第一反应是加资源但我们没有立刻操作而是先打开 Spark UI 看 Streaming 页面。6.2 排查步骤从批处理时长到单任务耗时Spark UI 的 Streaming 页面里能看到每个批次的调度延迟、处理时间、总延迟。当时一眼就看到了问题从某个时间点开始每个 Batch 的处理时间突然从 2 秒左右跳到了 7 秒以上后面的批次开始排队。接着点进那个异常的 Batch展开 Job 详情看到一个 stage 里某个 task 的运行时间异常接近 100 秒而其他 task 只需要几秒。到这一步基本可以断定不是整体资源不够而是数据倾斜——某些 key 的数据量远超平均线拖垮了单个分区上的计算任务。6.3 根因与修复热点 key 的两阶段聚合顺着 task 定位到具体算子发现是reduceByKeyAndWindow里某个省份的订单量因为促销活动暴增这个 key 所在分区的计算压力是其他分区的几十倍。之前也遇到过类似问题但没这次夸张。修复方案是对倾斜 key 做两阶段聚合先在 key 后面加一个随机后缀把它打散成多个子 key执行一次局部聚合再用去掉后缀的原始 key 做二次聚合。这样原本压在一个 task 上的计算量就分散到了多个 task 上。我们当时把随机后缀的范围设成了 10改造后同一个批次的处理时间从 7 秒降回了 2 秒左右。如果不用随机后缀而是直接把热点 key 单独分流也可以但要额外写路由逻辑通用性差一些。这个案例里最关键的一步是“先看数据再动资源”。如果当时没有排查就盲目增加 Executor倾斜问题解决不了只是用更多资源去掩盖单点瓶颈而且一旦资源加到一定程度成本上升了延迟还是压不下来反而更难定位。希望你在遇到类似告警时能想起这个思路看 Batch 处理时间、看单个 task 耗时、定位热点 key最后再决定是加资源还是改逻辑。6.4 一个值得记住的排查顺序把整个排查链路整理成清单下次可以直接照着做打开 Spark UI Streaming 页面看哪些批次处理时间明显放大。点进异常批次看是哪个 Job、哪个 stage 耗时异常。对比同一 stage 内各个 task 的耗时差异选出长尾 task。查看长尾 task 处理的数据分布定位热点 key。根据热点类型选择两阶段聚合或分流改造改造后观察下一批处理时间是否回落。如果任务本身没有倾斜再考虑资源、GC、网络延迟等问题。这套链路看起来笨实话说确实是最省时间的。我在不同项目里反复用过十几次线上延迟问题八成都能在这一条链路里找到根因。尤其是“同一 stage 里 task 耗时不均衡”这个信号几乎可以直接指向数据倾斜不需要再去翻各种监控面板。我个人在写完这一段后最大的体会是搞实时计算不要一上来就追新框架先把正在用的框架吃透。Kafka Spark Streaming 这套技术栈只要把分区、偏移量、背压、幂等输出这几个关键点拿捏住绝大多数业务场景都扛得下来。最后再分享一个小技巧每次提交偏移量之前把当前批次的 offsetRanges 按分区打印到日志里。这个动作对排查重复消费和乱序消费特别有用等你要对账数据的时候你会发现这些日志比任何监控面板都靠谱。