1. 先把 Spark Streaming 的定位摆正它到底是什么、什么时候不该用它Spark Streaming 是大数据开发里一个绕不过去的名字但它同时也是近几年被误解最深的一个组件。很多人第一次接触实时计算上手的就是它写着写着发现延迟下不去、数据对不上、重启之后状态全丢于是转头就给它贴上过时的标签。我的看法是它没有过时只是它的能力边界被大量教程模糊掉了导致很多人在错误的场景里用了正确的工具最后得出错误的结论。先把最基础的事实说清楚。Spark Streaming 是 Spark 核心 API 之上的一个流式计算扩展库它把连续到达的数据流切分成一个个小批次每个批次交给 Spark 引擎当作一个普通的 RDD 作业去执行。这个设计带来的直接后果是你写流处理代码的方式和写离线批处理代码几乎一模一样map、filter、reduceByKey、join这些算子在两边是通用的。对于一支已经有 Spark 离线经验的团队来说迁移成本极低这是它当年能迅速铺开的核心原因。但它能解决的问题是有明确边界的。它擅长的是秒级到分钟级的准实时场景比如分钟级的业务指标汇总、日志的准实时清洗入仓、订单状态的近实时同步、风控规则的准实时打标。它不擅长的是毫秒级的硬实时也不擅长事件驱动、单条触发的逻辑。你要是拿它去做一个要求 100 毫秒内响应的告警系统无论怎么调优都会很别扭因为微批的调度开销本身就摆在那里。这里还有一件必须提前说清楚的事从 Spark 2.x 后期开始官方主推的流处理方案已经换成了 Structured StreamingSpark Streaming也就是常说的 DStream API进入了维护状态新特性基本不再往里加。但这不意味着存量项目要立刻推倒重来。我接触过的生产环境里仍然有大量基于 DStream 的作业在稳定跑着尤其是那些逻辑复杂、状态管理重度依赖updateStateByKey、并且团队对 DataSet/DataFrame API 不熟的老链路。理性做法是新项目优先评估 Structured Streaming老项目按需维护真正需要重写的时候再重写而不是为了用新东西而制造一次高风险迁移。1.1 微批不是真流理解这条前提能省掉一半返工微批micro-batch这个词听起来很技术其实打个比方就明白了。真正的流处理像水龙头水一直在流你随时伸手都能接到水微批像用一个杯子去接水接满一杯处理一杯杯子和杯子之间有明确的边界。这个边界就是batch interval也就是你创建StreamingContext时传入的那个时间参数。设成 5 秒就意味着每 5 秒生成一个批次这个批次里装的是这 5 秒内到达的所有数据。你的处理逻辑看到的是一批数据而不是一条数据。理解这一点之后很多现象就顺理成章了。为什么延迟最小也就接近一个 batch interval因为数据要等这一批攒够了才会被处理。为什么批次处理时间必须小于 batch interval因为如果一批要跑 8 秒而间隔只有 5 秒批次就会排队堆积调度延迟Scheduling Delay一路往上涨最后内存扛不住直接崩。为什么窗口操作的长度必须是 batch interval 的整数倍因为窗口的滑动是以批为单位计算的不是以时间为单位计算的。我见过太多人卡在这一步他们默认流处理就是来一条处理一条于是写出每来一条数据就查一次数据库、发一次 HTTP 请求的逻辑。在微批模型下这种写法会让单批次的处理时间爆炸因为一个批次里可能有几万条数据你等于在一个批次里串行发了几万次请求。提示判断一个逻辑能不能塞进流处理先问自己一句——这段代码在一个批次内被执行 N 次N 等于批内记录数会不会出事会出事就必须改成批量化操作。1.2 什么场景适合什么场景应该换方案我把判断标准整理成一张表实际选型的时候对着看比凭感觉靠谱。场景特征建议方案理由秒级到分钟级延迟逻辑复杂需要和离线共用代码Spark Streaming算子通用团队上手快生态成熟需要毫秒级响应、单事件触发其他流式框架微批的调度开销无法绕过数据源是 Kafka需要精确一次语义Spark Streaming 或 Structured StreamingDStream 需要自己管 offset 和幂等Structured Streaming 内建支持更好重度依赖复杂状态、会话切分优先评估 Structured StreamingDStream 的状态 API 更笨重状态膨胀难治理流批一体同一套逻辑跑离线和实时Structured StreamingDStream 的 DataFrame 支持有限存量 DStream 作业稳定运行、逻辑复杂维持现状做好监控重写风险大于收益表格里最后一行是我个人的真实态度。我见过不止一次为了技术升级而重写流作业的项目最后因为业务逻辑里埋了太多历史遗留的特殊分支重写版本上线后对账一直不平折腾了两个月又回滚。技术选型的第一原则永远是能不能稳定交付业务价值而不是这个东西是不是最新的。2. 运行骨架拆解一个 batch 从产生到跑完经历了什么写流作业出问题的时候如果只知道算子怎么调排查会很盲目。真正高效的排查方式是脑子里有一张运行时结构图数据从哪进来、谁在调度、谁在执行、卡在哪一环。这一节我把这条链路拆开讲。Spark Streaming 作业启动后Driver 端会构建一个叫DStreamGraph的对象它是所有 DStream 转换关系的图结构。这个图在启动时就被固定下来之后每个批次都是拿这张图去生成一份 RDD 的 DAG。负责这件事的是JobScheduler它内部有一个定时器按照 batch interval 周期性触发generateJobs把 DStreamGraph 转换成一组 RDD 作业再通过JobExecutor提交给 SparkContext 执行。关键点在于JobExecutor的线程池大小。默认只有一个线程对应参数spark.streaming.concurrentJobs默认值 1也就是说同一个批次内的多个输出操作是串行提交的。如果你的代码里写了三个foreachRDD它们会排队执行总耗时是三段之和。有人为了提速把这个参数调大结果发现作业之间争抢资源、批次反而更乱这就是没搞清楚它是并发提交而不是并发加速。2.1 Driver 端的 DStreamGraph 与 JobSchedulerDStreamGraph有一个容易被忽视的特性它在作业启动那一刻就被确定了运行期间不会因为你改了代码而动态变化。这个特性直接影响后面要讲的 checkpoint 恢复问题——因为 checkpoint 里存的就是这个图结构图和代码一旦对不上反序列化就会失败。JobScheduler的调度逻辑里有一个叫batchDuration的东西它决定了时间片的切分。每个时间片结束时调度器会检查这个时间片内是否有数据、是否需要生成作业。生成出来的作业对象会带上一个time字段这个时间戳是批次的时间标签不是数据本身的处理时间。你在窗口操作、状态更新里看到的所有时间都是这个批次时间。这里有个实操细节值得记一下如果某个批次没有任何数据JobScheduler默认仍然会生成作业只是 RDD 为空这会带来一些无意义的空跑。可以通过spark.streaming.receiver.writeAheadLog.enable相关的配置配合空批次判断来减少开销但更简单的做法是在foreachRDD里先if (!rdd.isEmpty())再执行真正的逻辑避免空批次去连数据库、查外部接口。另一个常见坑是时钟问题。批次时间依赖 Driver 所在机器的系统时钟如果这台机器和 Kafka 集群、或者和其他节点的时间偏差较大会出现批次时间错乱、offset 定位异常之类的怪现象。生产环境务必保证 NTP 同步这不是最佳实践这是硬要求。2.2 Receiver 模式与 Direct 模式的本质区别和 Kafka 对接是绝大多数人的真实场景而这里恰好是坑最密集的地方。早期的方式叫 Receiver 模式Spark 在 Executor 上起一个 Receiver 线程持续从 Kafka 拉数据拉到的数据先写进 Spark 的块管理器BlockManager同时如果开了 WAL还会先写一份到可靠的分布式存储。然后作业再从这些块里读数据。这个模式的问题非常明显。第一Receiver 线程会长期占用一个 Executor 核心如果你的资源是local[1]或者每个 Executor 只给 1 个 core那个 core 被 Receiver 占了就没有 core 去跑处理逻辑了作业会像卡死一样不动。这是新手最常踩的坑表现为作业启动了但一条数据都不处理。第二WAL 虽然提升了可靠性但每次都要多写一份到外部存储吞吐直接砍掉一截。第三Receiver 和 Kafka 分区的对应关系不直观扩分区之后消费均衡性很难保证。Direct 模式也叫直连模式对应spark-streaming-kafka-0-10这套集成彻底换了思路不再有常驻的 ReceiverDriver 在每个批次开始时直接向 Kafka 查询这批要消费的每个分区的起止 offset然后把这个范围下发给各个 Task由 Task 自己去 Kafka 拉数据。RDD 的分区数天然等于 Kafka 的分区数或者你指定的范围数消费并行度和 Kafka 分区一一对应扩分区就等于扩并行度。这个变化带来的连锁好处是offset 变得可编程了。你能在代码里精确拿到每个分区消费到了哪能自己决定什么时候提交、提交到哪。但反过来责任也转移到了你身上——Direct 模式不会帮你自动提交 offset你必须自己写这段逻辑写错就是丢数据或者重复消费。注意Receiver 模式下 RDD 的分区数由spark.streaming.blockInterval默认 200ms决定一个批次内分区数约等于 batchInterval 除以 blockIntervalDirect 模式下分区数等于 Kafka 分区数。这两种模式的分区逻辑完全不同调优时不要混用经验。2.3 并行度到底由谁决定并行度这个问题问的人最多答清楚的人最少。我把它拆成三层来看。第一层是数据源并行度。Direct 模式下就是 Kafka 分区数这是你能达到的并行度上限。如果你有 3 个分区无论给多少 Executor同一时刻最多只有 3 个任务在消费。想要更高吞吐唯一的办法是扩分区。第二层是计算并行度。经过reduceByKey、repartition这类会产生 shuffle 的算子之后分区数会变后续算子的并行度跟着变。reduceByKey不指定分区数时默认用spark.default.parallelism这个值如果没配往往会比 Kafka 分区数小很多形成瓶颈。第三层是资源并行度。Executor 数量乘以每个 Executor 的核数决定了同一时刻最多能跑多少个 Task。资源不够任务排队资源过多任务空转。这三层里最容易被忽略的是第一层和第三层的匹配。举个例子Kafka 有 12 个分区你给了 4 个 Executor、每个 4 核总共 16 个核理论上刚好够。但实际会有 4 个核空着因为一个批次最多只能起 12 个消费任务。反过来如果 Kafka 只有 3 个分区你给 20 个核也是白给CPU 利用率永远上不去。所以我看作业的第一眼永远是先对一下分区数和总核数这两个数。3. 核心算子实操从无状态到有状态再到窗口算子这部分官方文档列得很全但文档不会告诉你什么时候该用哪个。我按实际会用到的顺序过一遍重点讲取舍逻辑。3.1 无状态转换transform 与 foreachRDD 才是主力map、flatMap、filter这些算子和普通 RDD 一模一样没什么好讲的。真正需要重点说清楚的是transform和foreachRDD因为它们是打通流处理和外部世界的两扇门。transform的作用是把一个 DStream 转成另一个 DStream但转换函数接收的是每个批次的 RDD返回的也是 RDD。这意味着你可以在里面做任何 RDD 层面的操作包括和外部静态数据做 join。比如你要给每条订单打上商户的行业标签商户表不大几十万行就没必要每条去查一次外部存储直接把商户表作为广播变量在transform里和批次 RDD 做 join效率高得多。foreachRDD是所有输出的终点它接收批次 RDD 但不返回新流专门用来做把数据推出去这件事写数据库、写消息队列、落文件、发请求。这里有个几乎所有人第一次都会犯的错误——在foreachRDD外面创建连接在里面使用。// 错误示范连接在 Driver 端创建会被序列化下发必然报 NotSerializableException val conn DriverManager.getConnection(url) dstream.foreachRDD { rdd rdd.foreach { record val stmt conn.prepareStatement(insert into t values (?)) // ... } }正确做法是按分区创建连接并且在分区内部复用dstream.foreachRDD { rdd rdd.foreachPartition { partitionOfRecords if (partitionOfRecords.nonEmpty) { val conn DriverManager.getConnection(url) try { val stmt conn.prepareStatement(insert into t values (?, ?)) partitionOfRecords.foreach { record stmt.setString(1, record._1) stmt.setString(2, record._2) stmt.addBatch() } stmt.executeBatch() } finally { conn.close() } } } }这段代码的价值在于三点连接按分区创建数量可控最多等于分区数不会像按记录创建那样打爆数据库连接池批量提交把 N 次网络往返压缩成 1 次finally保证连接释放避免长时间运行后连接泄漏。我调过的性能问题里至少有三分之一是每条记录建一次连接造成的改完直接提升一个数量级。3.2 窗口操作的正确打开方式窗口操作是用来算最近一段时间内的统计量的比如最近 10 分钟每个品类的成交额。API 形式是window(windowLength, slideInterval)两个参数都必须是 batch interval 的整数倍否则启动时直接报错。这里必须理解窗口的计算代价。假设 batch interval 是 5 秒你要算最近 60 秒的指标滑动步长也是 5 秒。最朴素的实现是每 5 秒对最近 60 秒12 个批次的全部数据重新算一遍。如果你的流每秒有 5000 条一次要扫 12 个批次 30 万条数据5 秒一次压力不小。reduceByKeyAndWindow提供了一个优化版本多传一个逆函数参数。它的思路是增量化新窗口的结果 老窗口的结果 新进入批次的数据 - 移出窗口批次的数据。这样每次只需要算两个批次而不是 12 个批次。val windowed pairs.reduceByKeyAndWindow( (a: Long, b: Long) a b, // 新数据进入窗口时的聚合函数 (a: Long, b: Long) a - b, // 数据移出窗口时的逆函数 Seconds(60), // 窗口长度 Seconds(5) // 滑动步长 )逆函数这个机制有明确的前提聚合操作必须是可逆的。加减法可以求最大值不可以求去重后的计数也不可以。如果业务逻辑用不了逆函数就只能用全量重算的版本这时候要特别注意控制窗口长度、控制单批数据量必要时把结果按照业务主键预聚合之后再做窗口把参与窗口计算的数据量压下去。注意使用带逆函数的窗口时如果某个 key 在窗口内所有数据都移出后计数变成 0这个 key 会从结果里消失。如果下游依赖key 存在但值为 0这个语义需要额外处理比如用一个特殊标记的filterFunc这个重载版本支持filterFunc参数或者在写入下游时做兜底。这个坑我在对接下游报表的时候踩过指标忽有忽无查了很久才发现是计数归零被自动移除了。3.3 状态管理updateStateByKey 与 mapWithState 的取舍状态管理是流处理从能用到好用的分水岭。典型的业务需求是给每个设备维护一个累计在线时长或者给每个用户维护一个会话状态。updateStateByKey允许你传入一个函数参数是当前批次这个 key 的所有 value和上一个批次的 state返回新的 state。def updateFunc(newValues: Seq[Long], state: Option[Long]): Option[Long] { val current newValues.sum val previous state.getOrElse(0L) Some(current previous) } val cumulative pairs.updateStateByKey(updateFunc)这个 API 有两个必须知道的特性。第一它强制要求开启 checkpoint否则启动就报错因为状态需要持久化到可靠存储才能容错。第二每一次更新都会对全量的 key 做一次 shuffle 和全量状态扫描状态越多批次处理时间越长。状态量大到一定程度就会出现越跑越慢最后跑不动的情况这是典型的性能瓶颈。mapWithState是后来引入的替代方案它把状态封装在State对象里支持超时机制内部实现基于MapWithStateRDD只对发生变化的分区做计算性能比updateStateByKey好很多官方给出的数字是数量级级别的提升。val mappingFunc (userId: String, amount: Option[Double], state: State[Double]) { val total state.getOption().getOrElse(0.0) amount.getOrElse(0.0) state.update(total) (userId, total) } val spec StateSpec.function(mappingFunc).timeout(Seconds(1800)) val stateDStream pairs.mapWithState(spec)timeout这个参数很有价值超过指定时间没有更新的 key会自动从状态里清除。这一条直接解决了状态无限膨胀的问题。原理上timeout控制的是状态是否存在业务上如果要求用户离线 30 分钟就关闭会话这个参数天然契合。我的建议很明确新写的有状态逻辑一律用mapWithState存量代码用updateStateByKey的如果状态量还在可控范围内不动它如果已经出现批处理时间持续上涨优先考虑迁移到mapWithState或者干脆把状态挪到外部存储比如 Redis自己管理把 Spark 只当作计算引擎。3.4 攒批写库foreachRDD 里的连接管理前面讲了按分区建连接这里再补一个更细的点批次内部再攒批。有些场景单个分区的记录数很多几万条一次性executeBatch也可能造成数据库压力过大或者超时报错。可以在分区内部再按固定大小切片比如每 500 条提交一次。rdd.foreachPartition { iter val conn DriverManager.getConnection(url) try { var count 0 val stmt conn.prepareStatement(sql) iter.foreach { record stmt.setString(1, record._1) stmt.addBatch() count 1 if (count % 500 0) stmt.executeBatch() } if (count % 500 ! 0) stmt.executeBatch() } finally { conn.close() } }这个 500 不是随便定的。它取决于三个因素单条记录的字段宽度、数据库max_allowed_packet之类的限制、以及单次批量提交的耗时上限。我的经验是从 200 开始试逐步往上加直到单次提交耗时稳定在 1 秒以内再往上就没有收益了。4. 和 Kafka 对接的完整链路offset 到底该放在哪Kafka 是流处理最常见的数据源而 offset 管理是这个组合里最需要想清楚的事。offset 管不好轻则重复消费重则数据丢失而且这类问题往往在作业重启之后才暴露属于典型的平时不响出事就是大事。4.1 Direct 模式的代码骨架先给一份可以直接跑起来的基础结构我尽量把每一步的意图标注清楚。val conf new SparkConf() .setAppName(order-stream) .set(spark.serializer, org.apache.spark.serializer.KryoSerializer) .set(spark.streaming.backpressure.enabled, true) .set(spark.streaming.kafka.maxRatePerPartition, 2000) val ssc new StreamingContext(conf, Seconds(5)) ssc.checkpoint(hdfs:///streaming-checkpoint/order-stream) val kafkaParams Map[String, Object]( bootstrap.servers - broker-1:9092,broker-2:9092, key.deserializer - classOf[StringDeserializer], value.deserializer - classOf[StringDeserializer], group.id - order-stream-group, auto.offset.reset - earliest, enable.auto.commit - (false: java.lang.Boolean) ) val topics Array(order-topic) val stream KafkaUtils.createDirectStream[String, String]( ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) )几个参数值得单独说。enable.auto.commit必须显式关掉交给 Spark 自己控制否则自动提交的时机和你处理数据的时机完全脱节一致性无从谈起。auto.offset.reset设为earliest还是latest取决于业务对第一次启动的期望——想从头补数据用earliest只处理新数据用latest。这个参数只在找不到已提交 offset时生效别指望它能解决 offset 越界的问题。LocationStrategies.PreferConsistent的意思是尽量把分区均匀分布到所有 Executor 上适合集群节点数足够的情况。如果某些节点和 Kafka 集群网络更近可以用PreferBrokers把消费任务调度到同一台机器上减少跨机流量。如果某个 Kafka 分区数据量明显偏大数据倾斜可以用PreferFixed手动指定映射关系把大分区单独放到资源充足的节点。4.2 offset 提交的三个方案对比这是我认为整个链路里最需要决策的地方我列一个表对比。方案实现方式一致性适用场景交给 Kafka 自动提交enable.auto.committrue最弱可能丢也可能重完全不推荐用于生产处理前提交拿到 offset 就commitAsync再处理数据会丢数据绝不允许处理后同步提交处理完成后commitSync再写下一条至少一次会重复大多数场景的默认选择处理后异步提交处理完成后commitAsync接受少量重复至少一次更轻量吞吐优先、下游可去重offset 存外部存储 幂等写处理结果和 offset 在同一事务/同一次写入中落库精确一次配合幂等对账严格、不能容忍重复这里最容易出的事故是处理前提交。表面上看代码逻辑很顺先告诉 Kafka 我消费到哪了然后再慢慢处理。但如果在处理过程中作业挂了重启之后 Kafka 认为你已经消费过了那批数据就永远不会再来了。数据静默丢失对账的时候才能发现。正确的顺序永远是先处理完再提交。拿到 offset 的方式是在foreachRDD里做类型转换stream.foreachRDD { rdd val offsetRanges rdd.asInstanceOf[HasOffsetRanges].offsetRanges // 先把业务逻辑做完 process(rdd) // 全部成功后再提交 stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges) }commitAsync是异步的性能好但如果提交失败Kafka 那边不知道下次启动会从旧的 offset 重新消费造成重复。commitSync更稳但会阻塞当前线程。我的取舍是如果下游写入是幂等的比如按主键 upsert用commitAsync换取吞吐如果下游是追加写且不能去重用同步提交牺牲一点性能换确定性。还有一点必须强调HasOffsetRanges的类型转换只能在foreachRDD的最外层做一旦经过map、filter这类算子RDD 就变成了普通的MapPartitionsRDD转换会直接抛ClassCastException。这个错误信息很不直观很多人第一次遇到会以为是版本不兼容。4.3 端到端一致性怎么拼出来精确一次这个词被用得很泛滥实际上它不是一个开关而是几层机制拼起来的。在 Spark Streaming Kafka 这个组合里我的理解是这样的第一层是消费端不丢。这靠处理完再提交 offset保证也就是至少一次的语义Kafka 那边保证数据不会在你没处理完的时候被标记为已消费。第二层是处理端可重放。这靠 checkpoint 和 offset 外部化保证重启后能从正确的位置继续而不是从头或者从某个随机位置开始。第三层是写入端幂等。这一层最容易漏但恰恰是关键。因为前两层只能保证至少一次意味着重启之后必然有一部分数据会被重复处理。要消除这个重复的影响只能靠写入端自己去重用业务主键做 upsert或者用一个已处理 offset 记录表配合事务写入。一个我实际用过的组合是这样的业务表用订单号作为唯一键写入时用INSERT ... ON DUPLICATE KEY UPDATE同时把每个分区的 offset 和处理时间写进一张独立的流水表两边放在同一个数据库事务里。这样即使重复消费业务数据不会重复而且一旦对账不平可以直接查流水表定位到具体是哪个 offset 区间出的问题排查效率高很多。注意如果选择把 offset 存外部存储比如 MySQL、Redis、ZooKeeper那么 checkpoint 里的 offset 信息就变成了冗余甚至是干扰。重启时必须明确以外部存储为准不能让两套机制同时生效否则会出现checkpoint 恢复到了旧 offset 但外部存储是新 offset的冲突。我自己的习惯是一旦用了外部 offset 存储就直接关掉 offset 相关的 checkpoint 依赖ssc.checkpoint只用来存状态。5. Checkpoint 与容错用得好是保险用不好是负债Checkpoint 在 Spark Streaming 里的地位有点尴尬。文档里到处都在说要开但真正踩过坑的人对它又爱又怕。我把它拆成它到底存了什么和什么时候该用它两部分来讲。5.1 Checkpoint 到底存了什么ssc.checkpoint(path)指定一个可靠存储目录HDFS、对象存储之类绝对不能用本地磁盘因为 Driver 重启可能换机器。这个目录里存的东西分三类第一类是元数据包括StreamingContext的配置、DStreamGraph的结构、未完成的批次信息。这一部分是 checkpoint 恢复的核心也是各种恢复失败的根源。第二类是状态数据updateStateByKey、mapWithState维护的状态 RDD 会定期写进来。这是真正有价值的部分没有它重启后所有累计状态清零。第三类是未处理的批次数据当开启了 WALspark.streaming.receiver.writeAheadLog.enable时Receiver 收到的数据会先写到这里再处理。这一块在 Direct 模式下不适用因为 Direct 模式的容错靠重放 Kafka 数据不需要 WAL。理解了这三类很多事情就说得通了。为什么开了 checkpoint 作业会变慢因为要周期性写元数据和状态。为什么改了代码之后恢复会失败因为元数据里的DStreamGraph和你的新代码对不上。为什么状态特别大的作业 checkpoint 很慢因为状态数据本身就很占空间。5.2 优雅停止与重启恢复的实操流程停机这件事看起来简单实际上处理不好就是一次数据事故。最忌讳的操作是直接kill -9。这种情况下正在处理的批次会被硬生生打断如果 offset 刚好提交了但数据没写完就是丢失如果数据写了但 offset 没提交就是重复。正确的做法是走优雅停止流程。第一种方式是配置spark.streaming.stopGracefullyOnShutdowntrue然后向 Driver 发送SIGTERM普通的kill不是kill -9Spark 会在收到信号后停止接收新数据把当前批次处理完再关闭上下文。第二种方式是在代码里注册关闭钩子sys.addShutdownHook { ssc.stop(stopSparkContext true, stopGracefully true) }这两个参数的含义要清楚stopGracefully决定是否等待当前批次处理完成stopSparkContext决定是否连带关闭 Spark 上下文。如果同一个 JVM 里还有别的 Spark 作业第二个参数要设成false否则会把别人的上下文也关掉。重启的流程我整理成固定的四步确认旧进程已经完全退出检查 YARN 或者集群管理器上的应用状态确认最后一轮的 offset 已经正确提交查外部 offset 表或者 Kafka 的消费位点用同一个 checkpoint 目录启动观察启动后的第一批数据量和 offset 是否是接续的而不是跳出很远。第三步里有个细节值得注意如果 checkpoint 目录里存在不完整的写入前面被强杀留下的启动时可能报反序列化错误。这时候的处理办法不是把 checkpoint 删掉重来会丢状态而是先把目录备份一份再尝试启动如果确实不行才考虑放弃 checkpoint改用外部 offset 从某个已知位置重放。5.3 代码升级后 checkpoint 不兼容怎么办这是最现实的痛点。业务要加个字段、改个过滤条件代码一改用老 checkpoint 启动就报ClassNotFoundException或者InvalidClassException。原因就是前面说的checkpoint 里序列化了整个DStreamGraph包括你的匿名函数类类的结构和包名一变反序列化就失败。我见过的处理方式有这么几种按推荐程度排第一种不依赖 checkpoint 恢复作业只用来存状态和无状态恢复。具体做法是把 offset 完全交给外部存储管理重启时从外部存储读取 offset从头构建 DStreamGraphcheckpoint 目录换一个新的。这样代码升级完全不受影响代价是状态需要额外处理要么接受状态丢失重算要么把状态也外部化到 Redis 之类的地方。第二种用getOrCreate做兼容。StreamingContext.getOrCreate(checkpointPath, creatingFunc)的逻辑是如果 checkpoint 存在就从中恢复不存在就调用creatingFunc新建。它解决的是首次启动的问题解决不了代码变了的问题但它能让你在新建分支里统一初始化逻辑代码结构更干净。第三种升级前主动放弃 checkpoint。把老的 checkpoint 目录归档代码里改用一个全新的路径从外部 offset 位置重新消费。这是最干脆的做法前提是业务能接受状态重建的成本。提示从工程实践看凡是需要频繁迭代逻辑的流作业我都不建议重度依赖 checkpoint 恢复元数据。把 checkpoint 的定位收窄到存状态offset 交给外部存储代码升级的自由度会大很多。这个结论是我被反序列化报错折磨过三次之后才定下来的。6. 性能调优与线上故障排查调优这块我不想堆参数表因为参数表网上到处都是但照着抄往往没效果。真正有用的是搞清楚每个参数在解决什么问题以及怎么判断问题出在哪。6.1 批间隔、反压与限速的三个旋钮第一个旋钮是batch interval。它决定了延迟的下限也决定了每个批次的数据量。设得太小比如 1 秒批次调度开销占比会很高而且每个批次的数据太少各种初始化的成本摊不平设得太大比如 60 秒延迟上去了而且单批数据量大容易触发内存问题。我的经验值数据量在每秒几千到几万条的规模5 到 10 秒是比较舒服的区间。判断标准很直接——看 Spark UI 里的 Processing Time如果它长期稳定在 batch interval 的 50% 以下说明这个间隔是合适的如果经常超过 80%说明处理不过来要么调大间隔要么优化处理逻辑。第二个旋钮是反压backpressure。开启方式很简单spark.streaming.backpressure.enabledtrue它的作用是Spark 根据上一批次的处理速度动态调整下一批次拉取的数据量。处理快就多拉处理慢就少拉。这个机制能有效防止启动瞬间拉爆内存的情况——作业刚启动时Kafka 里积压了大量数据如果没有反压第一批就会把积压全部拉下来直接 OOM。反压的初始速率由一个内部估算值决定前几个批次可能不太准需要跑一段时间才能稳定。如果发现反压调整得太保守、吞吐上不去可以配合spark.streaming.backpressure.initialRate给一个合理的起始值。第三个旋钮是硬限速spark.streaming.kafka.maxRatePerPartition单位是每个分区每秒最多消费多少条。它和反压的区别是反压是动态的、自适应的限速是静态的、硬性的。通常两个一起用反压管自适应限速兜底防止极端情况。举个例子算一下。Kafka 有 8 个分区每个分区限速 2000 条每秒batch interval 是 5 秒那么单批次最多处理 8 × 2000 × 5 80000 条。这个数字要和你下游的承载能力对得上。如果下游数据库每秒只能写 5000 条那 80000 条一批根本写不完必须把限速降下来或者把 batch interval 拉长让压力平摊。6.2 序列化与内存序列化这块几乎是一个不开 Kryo 就是浪费的状态。默认的 Java 序列化又慢又占空间换成 Kryo 通常能省 30% 到 50% 的内存和网络传输量。spark.serializerorg.apache.spark.serializer.KryoSerializer spark.kryo.registrationRequiredfalse如果你的自定义类比较多建议显式注册spark.kryo.classesToRegistercom.example.OrderEvent,com.example.UserState注册之后 Kryo 不需要在每条记录里写类名体积进一步压缩。registrationRequired设成false是图省事设成true更严格会强制所有类都注册能提前暴露某个类没注册的问题但开发阶段会比较烦。我的做法是先在开发环境设true跑通把该注册的类都找出来生产环境再决定是否放开。内存相关的还有几个点。spark.rdd.compresstrue可以对缓存的 RDD 做压缩用 CPU 换内存在内存吃紧的时候很有用。spark.memory.fraction控制执行内存和存储内存的比例默认 0.6如果作业里有大量缓存操作可以适当调高如果是纯计算型没有缓存可以调低给执行内存腾地方。另外状态特别大的作业要关注 GC。如果日志里频繁出现 Full GC或者 UI 里 GC Time 占比超过 10%首先要排查的是不是状态无限增长其次才是调 GC 参数。-XX:UseG1GC在大多数场景下比 CMS 表现更稳尤其是堆比较大的情况。6.3 常见问题速查表下面这张表是我这些年攒下来的基本覆盖了 80% 的线上问题。现象常见原因排查方向作业启动了但不处理数据Receiver 模式下local[1]被 Receiver 占满改用 Direct 模式或把核数提到 2 以上调度延迟持续上涨批次堆积单批处理时间大于 batch interval看 Processing Time优化算子或调大间隔Task not serializable闭包里引用了不可序列化的对象如连接、客户端把对象创建挪进foreachPartitionClassCastException在HasOffsetRanges处在map/filter之后才做类型转换转换必须放在foreachRDD第一行重启后大量重复消费offset 提交失败或提交时机不对检查提交逻辑下游加幂等重启后数据丢失处理前就提交了 offset改为处理完成后提交checkpoint 恢复报反序列化错误代码变更导致类结构不匹配换 checkpoint 目录从外部 offset 重放状态越跑越大最终 OOMupdateStateByKey无超时机制迁移到mapWithState并设置 timeoutKafka 报OffsetOutOfRangeExceptionoffset 超出 Kafka 保留范围检查retention.ms调大保留时间或调整 reset 策略单批次数据倾斜个别任务跑不完某些 Kafka 分区数据量不均检查分区分配必要时用PreferFixed手动映射数据库连接池被打满每条记录建一次连接改为按分区建连接批内复用这张表里前三行的出现频率明显高于其他。尤其是第一行我见过太多人在本地测试的时候用local[*]一切正常部署到只有 1 个核的环境上就卡死查了半天以为代码有问题。6.4 几个真实踩坑现场说两个印象比较深的例子都是从看起来没问题到线上出问题的典型。第一个是空批次导致的连接风暴。有个作业的逻辑是每批都去连一下配置中心拉最新的规则。平时好好的某天凌晨业务低峰Kafka 里一条数据都没有但作业还在按 5 秒一次的频率跑每批都去连配置中心。结果配置中心的连接数被打满连带影响了其他系统。原因就在于没有判断rdd.isEmpty()空批次照样走完了整套逻辑。加上判空之后问题消失。第二个是窗口长度和滑动步长配错。有人想算最近 1 小时的指标window(Seconds(3600), Seconds(60))batch interval 是 5 秒。看起来没问题但实际上每次窗口计算要聚合 720 个批次而且是全量重算。线上跑了两天状态 RDD 越来越大Full GC 越来越频繁最后 Driver OOM。改成增量窗口加预聚合之后单批处理时间从 40 多秒降到了 2 秒以内。第三个是offset 提交顺序和写入顺序不一致导致的对账不平。代码里先commitAsync再写数据库理论上中间只有几毫秒的窗口期但就是在某次机房网络抖动的时候恰好命中了那一批数据永远丢了。这个问题的可怕之处在于它发生的概率极低可能几个月才碰上一次但只要碰上排查成本极高因为日志上看不出任何异常。改法是把这个顺序反过来并且加了 offset 流水表做对账。我个人在这些事上得到的体会是流处理作业的稳定性和代码写得多漂亮关系不大关键是把几件事固定成纪律offset 先处理后提交、外部写入必须幂等、任何循环里都要考虑批量化、checkpoint 不要承担它承担不了的责任。这些纪律在写第一行代码的时候就定下来比事后调优有用得多。还有一个我自己一直在用的笨办法每个流作业上线前强制做一次断电演练。就是在压测环境里故意在作业处理数据的过程中强杀进程然后重启看数据有没有丢、有没有重复、状态对不对。这个过程很枯燥但它能提前暴露 90% 的容错问题。我见过太多作业在正常情况下跑三个月没事一遇到集群重启或者节点故障就露馅与其到时候手忙脚乱不如上线前花两个小时把这条路走一遍。