Spark Streaming生产调优：微批、状态、位点与故障恢复

第一次把 Spark Streaming 任务推到生产环境凌晨两点被手机报警震醒——批处理时间从 8 秒一路涨到 40 秒任务队列里堆了十几个未处理的 batchKafka 消费位点越落越远。翻日志发现不是集群资源不够也不是数据量突增而是一个foreachRDD里藏着一次阻塞式的外部调用。这种问题在文档里几乎找不到只有真正在生产上跑过流任务的人才会踩到。这篇想聊的就是这类东西。Spark Streaming 这套微批处理框架入门门槛看着不高StreamingContext、DStream、几个算子几十行代码就能跑起来一个实时统计。但真正决定一个流任务能不能稳定跑上一年半载的是那些藏在批间隔、位点管理、状态存储和反压机制里的细节。下面按我实际踩过的顺序从它到底怎么算、算子怎么用、数据怎么进、状态怎么管、性能怎么调、故障怎么恢复一直到该不该继续用它一条线索捋下来。适合已经写过几个 Demo、正准备把任务推上生产的人也适合跑了一段时间但一直被延迟和重复数据困扰的人。1. 微批模型的本质Spark Streaming 到底在算什么很多人对 Spark Streaming 的第一印象是实时但它的英文名里就写着 Streaming底层却是彻头彻尾的批处理。理解这一点后面所有的调优和踩坑才有解释的起点。它把一条连续不断的数据流按时间切成一个个固定长度的小块每一块交给 Spark 引擎当做一个普通 RDD 去算。这个切分动作叫微批micro-batch切分的间隔就是批间隔batch interval。1.1 一个批间隔里究竟发生了哪几件事设成 5 秒的批间隔意味着每 5 秒调度器会生成一个新的 RDD把它塞进一个叫 JobScheduler 的队列里然后由底层的 SparkContext 提交成 Job 去执行。这里有个关键点串行。同一时刻默认只有一个 Job 在跑spark.streaming.concurrentJobs 这个参数不调后面的批只能排队等前面的算完。这就解释了一个很常见的现象——你把批间隔从 5 秒改成 1 秒吞吐没上去延迟反而更糟了。因为每批的数据量变小了但每批的固定开销任务调度、序列化、shuffle 元数据、GC几乎不变。原来 5 秒处理 50 万条要 3 秒现在 1 秒处理 10 万条可能要 2 秒一秒一批根本来不及队列瞬间堆积。整个链条的公式很简单实际端到端延迟 ≈ 排队批数 × 批间隔 单批处理耗时单批处理耗时必须稳定小于批间隔否则排队批数只增不减这就是所谓的雪崩式积压。所以批间隔不是一个我想多实时的愿望参数而是一个我的集群能多快算完的容量参数。1.2 批间隔怎么定从处理能力倒推而不是从需求正推我见过太多人先定 1 秒然后发现跑不动再加机器。合理的顺序反过来先拿一天的峰值数据量算出每秒钟的条数再估算单条处理成本然后倒推。举个例子某条业务线峰值 QPS 是 6 万条每条记录约 800 字节做的是解析、维表关联、聚合三步。压测下来单个 CPU 核每秒能处理约 2500 条这个数字只能靠实测不同算子差异极大。集群可用核心数 48那么理论处理能力是 12 万条/秒。留 50% 的余量实际按 6 万条/秒算刚好打满。这种情况下批间隔定 5 秒每批 30 万条处理时间大约 2.5 秒稳定有余量如果定 1 秒每批 6 万条理论上 0.5 秒能算完但每批的固定开销往往就要 0.3 到 0.5 秒一抖动就超时。所以我的经验是批间隔不要低于 2 秒除非数据量极小且逻辑极简。业界大量稳定的 Spark Streaming 生产任务批间隔都在 5 到 60 秒之间。3 秒以下基本属于拿这个框架做它不擅长的事情。1.3 微批和真流处理的分水岭在哪里这里必须说清楚一个认知差异否则选型一定会错。微批模型的本质是离散化它有明确的批边界所以窗口计算天然对齐到批边界做 10 秒滚动窗口配 5 秒批间隔只能得到两个离散点你没法拿到逐条到达的精确时间状态是有版本快照的每批结束时状态落一次盘恢复时回放到最近的快照端到端延迟的下限就是一个批间隔再快也快不过它。而真正的流处理引擎是逐条驱动的延迟能到毫秒级窗口基于事件时间加水位推进。所以判断标准很直接业务能不能接受秒级延迟。做实时风控的毫秒级拦截、做实时竞价的价格计算微批模型天生不合适做分钟级的指标聚合、日志异常检测、订单状态流转监控它完全够用而且生态成熟、运维成本低。我在选型时有一条粗暴的线延迟要求在 1 秒以内别用微批1 到 10 秒可以接受但要谨慎10 秒以上随便用。这条线不绝对但能挡掉 80% 的选型错误。2. DStream 的算子谱系从逻辑描述到真正执行DStream 是个抽象概念它本身不存数据存的是一份计算描述——每个批间隔会基于这份描述生成一个 RDD。理解这一层才能明白为什么有些代码写完一点动静都没有为什么有些算子用完内存就爆。2.1 转换算子只画图行动算子才点着火DStream 的算子分两类和 RDD 的思路一致。map、filter、flatMap、reduceByKey、join、window 这些属于转换transformation它们只是往计算图上加一个节点不触发任何计算。只有 output 类算子——print、saveAsTextFiles、foreachRDD、saveAsHadoopFiles——才会真正注册一个输出操作Spark Streaming 才会为每个批生成 Job 去跑。这就带来一个新手很容易犯的错val lines ssc.socketTextStream(host, 9999) val words lines.flatMap(_.split( )) val counts words.map((_, 1)).reduceByKey(_ _) // 到这里什么都没发生一个任务都不会提交代码能编译、能启动、不报错日志里一切正常但就是没数据。补一句counts.print()或者counts.foreachRDD(...)才能跑起来。我在带新人的时候一定会强调DStream 代码不写 output 算子等于在纸上画了个流程图。还有一个隐性限制值得提前知道同一个 StreamingContext 里注册多个 output 算子默认它们是串行执行的。也就是说counts.print()和counts.saveAsTextFiles(...)会让同一批数据算两遍除非你开启spark.streaming.concurrentJobs。所以更好的做法是一个 output 算子搞定所有落地动作把打印和存储都放进同一个foreachRDD里。2.2 foreachRDD 里那几行最容易写错的代码foreachRDD是整个 Spark Streaming 里最容易踩坑的算子因为它在两个不同的上下文之间来回切换外面是 Driver里面是 Executor。dstream.foreachRDD { rdd // 这里是 Driver 端每批执行一次 val conn createConnection() // 错误示范连接在 Driver 建 rdd.foreach { record // 这里是 Executor 端每条记录执行一次 conn.insert(record) // 连接被序列化到 Executor废了 } conn.close() }这段代码的问题是一连串的。第一Driver 端创建的连接对象会被序列化后分发到各个 Executor绝大多数连接对象根本不可序列化直接抛NotSerializableException。第二就算侥幸能序列化连接在 Driver 建、在 Executor 用、在 Driver 关跨进程的句柄根本没有意义。第三rdd.foreach是逐条执行每条一次网络往返吞吐会低到令人发指。正确的写法是把连接的创建下沉到分区级别dstream.foreachRDD { rdd rdd.foreachPartition { partition if (partition.nonEmpty) { val conn createConnection() // 每个分区建一次 try { val batch partition.toList batch.grouped(500).foreach { chunk conn.batchInsert(chunk) // 批量提交 } } finally { conn.close() // 必须在 finally 里关 } } } }这里有几个细节值得单独拎出来。partition.nonEmpty的判断不能省空分区也会执行一次foreachPartition白白建一次连接。grouped(500)的批量大小要按目标存储的承受能力调太小浪费网络太大容易触发单次请求超时或者内存抖动。finally里的关闭更不能省Executor 上的资源泄漏不会立刻暴露要跑上几小时连接数打满才爆这时候排查成本极高。还有一个更隐蔽的问题foreachRDD里如果在 Driver 端做了rdd.collect()或者rdd.count()之类的行动操作等于每批多触发一次全量计算。曾经有个任务因为一句rdd.count()做日志埋点整体耗时翻了将近一倍去掉之后立刻恢复正常。2.3 窗口算子的对齐规则与性能代价窗口操作是流处理里最常用的能力window(windowLength, slideInterval)、reduceByKeyAndWindow、window配合transform都属于这一类。它的语义是每次计算覆盖最近一段时间的数据。关键在于两个参数的关系。windowLength是窗口长度slideInterval是滑动步长。默认slideInterval等于批间隔也就是说每批都算一次。如果窗口长度是 60 秒、批间隔 5 秒那么每次计算要覆盖 12 个 RDD。这 12 个 RDD 必须被保留在内存或磁盘里这就是窗口操作的内存代价。Spark Streaming 提供了一套优化方案reduceByKeyAndWindow的带invFunc版本可以用加新批、减旧批的增量方式代替全量重算val windowed pairs.reduceByKeyAndWindow( (a, b) a b, (a, b) a - b, 60, // 窗口长度 5 // 滑动步长 )这个优化能把计算量从每次扫 12 个 RDD降到每次只处理进出窗口的两个 RDD。但它成立的前提是聚合函数必须满足可逆性减法能还原加法才行。像求平均值、求去重后的 UV 这种不可逆操作就不能用这个版本只能老老实实全量算。用增量版本还有两个隐藏条件我在实际项目里都撞过一是必须开启 checkpoint因为窗口状态需要持久化才能恢复二是过滤条件要一致如果原始数据里有被 filter 掉的记录进出窗口的数量对不上累计结果会慢慢漂移。漂移这种事最恶心它不报错跑一天之后数字偏了 3%你不做对账根本发现不了。我自己现在的习惯是窗口长度不超过批间隔的 20 倍。超过这个比例状态占用和恢复时间都会变得不可控不如换个思路——把窗口聚合下沉到上游或者改用支持更好状态管理的方案。3. 数据接入与位点管理Kafka 到 Spark 那条链路的可控性流任务的数据源九成以上是消息队列剩下的才是 socket、文件、自定义 Receiver。这一节只讲 Kafka 这条路因为绝大多数生产问题和它相关。3.1 Receiver 模式为什么被淘汰早期的接入方式是基于 Receiver 的在 Executor 上起一个接收线程持续从 Kafka 拉数据攒成 block 存到 Spark 的内存里然后由 Spark 生成任务去处理这些 block。这个模型有几个致命问题。第一是数据丢失风险。接收线程把数据拉下来存到 Executor 内存同时向 Kafka 提交位点。如果在数据被处理完之前 Executor 挂了这部分数据就丢了。为了缓解引入了 WAL预写日志先把数据写一份到分布式存储再提交位点——代价是每条数据至少写两遍吞吐直接腰斩。第二是资源争抢。接收线程占着 Executor 的核真正做计算的核就少了。为了不拖慢计算往往要单独开一批 Executor 专门接收集群利用率很低。第三是分区与并行度脱钩。block 是按时间攒的一个 block 里可能混着多个分区的数据导致消费速率和 Kafka 分区数对不上想通过加分区来提速根本没用。现在的做法统一是Direct 模式也叫无 Receiver 模式。它不再起接收线程而是每个批间隔直接向 Kafka 查询这批要消费的分区从哪个位点读到哪个位点然后把这段区间包装成一个 RDD 交给 Spark。Kafka 的一个分区天然对应一个 RDD 分区一个 task并行度完全由 Kafka 分区数决定。这个模型的好处很直接数据不落地、不写 WAL、并行度可控、位点由用户自己管。代价是位点管理这件事从框架甩给了开发者这也成了后面最多坑的地方。3.2 位点到底该谁来管从自动提交到手动提交Direct 模式默认不提交位点你得自己来。最省事的写法是在foreachRDD里拿到OffsetRange然后提交val kafkaParams Map( bootstrap.servers - broker1:9092,broker2:9092, key.deserializer - classOf[StringDeserializer], value.deserializer - classOf[StringDeserializer], group.id - stream-group-01, auto.offset.reset - earliest, enable.auto.commit - false ) val stream KafkaUtils.createDirectStream[String, String]( ssc, PreferConsistent, Subscribe[String, String](topics, kafkaParams) ) stream.foreachRDD { rdd val offsetRanges rdd.asInstanceOf[HasOffsetRanges].offsetRanges // 先处理业务 process(rdd) // 处理完再提交位点 stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges) }顺序非常关键先处理完再提交位点。反过来写就成了 at-most-once处理失败时数据已经标记为消费过直接丢。这个顺序保证了 at-least-once代价是可能重复。commitAsync是异步的不会阻塞批处理。但它也有个坑如果下一批的位点提交比上一批先完成位点会倒退导致重复消费一大段。所以并发度高的场景我一般用同步的commitSync或者自己维护一个单调递增的提交逻辑。更稳妥的方案是把位点存到外部存储和业务结果放在一起提交。比如把位点写进业务库的同一张控制表和业务数据的写入放在同一个事务里。只有事务成功位点才前进事务失败位点不动重启后重放。这是端到端精确一次的基础后面第 6 节会展开。3.3 重复数据的现实处理去重表怎么设计只要用了 at-least-once重复就一定会发生。不是可能是一定——网络抖动、Executor 重启、批次重算任何一个都会带来重复。去重的核心思路是给每条记录一个业务主键 批次标识写库时做幂等。具体做法分几种方案实现方式适用场景代价主键冲突忽略INSERT IGNORE/ON CONFLICT DO NOTHING结果表有天然唯一键几乎为零首选覆盖写INSERT ... ON DUPLICATE KEY UPDATE需要更新而非忽略低去重表 存在性判断单独一张表记录已处理的 id主键不唯一的情况一次额外查询布隆过滤器内存位图判重数据量极大、容忍误判需要维护位图快照批次号跳过记录已处理的 batchId批次可重放且幂等最简单但粒度粗我最常用的是第一种因为大部分聚合结果表本来就有维度主键。如果实在没有我会造一个——把维度字段拼成字符串做哈希作为唯一键。哈希长度用 16 位十六进制足够碰撞概率在一亿量级下也还能接受而且比原始字段拼接省空间、比自增 id 好维护。有个细节容易被忽略去重判断必须在写入目标存储的同一个事务里完成不能先查后写。先查后写在高并发下会漏判两个批次同时查到不存在然后都插进去了。这种事在压测时基本测不出来只有生产上量之后才会零星出现。4. 状态计算与 checkpoint看不见的成本正在累积凡是跨批次的统计——累计 UV、会话聚合、连续异常计数——都需要状态。DStream 提供了updateStateByKey和mapWithState两个算子来维护状态用起来不复杂但背后的存储代价和恢复代价经常被严重低估。4.1 updateStateByKey 的状态膨胀是怎么发生的updateStateByKey的机制是每一批把所有 key 的当前状态和这一批的新值合并得到新的全量状态。注意全量两个字——每一批都要遍历并重写所有历史 key 的状态。假设你的任务是统计每个用户 ID 的累计访问次数跑了一个月累计有 3000 万个不同用户。那么每一批即使这一批只来了 100 条新记录框架也要把 3000 万个 key 的状态全集扫一遍、shuffle 一遍、写一遍。批处理时间会被这个固定的状态规模死死卡住跟当前批的数据量几乎无关。这就是典型的跑得越久越慢。第一天处理 1 秒第三十天处理 30 秒然后突然某天开始积压你怎么查数据量都查不出问题因为瓶颈在状态规模不在输入。解决方案是给状态加过期机制def updateFunc(newValues: Seq[Int], state: Option[Int]): Option[Int] { val current state.getOrElse(0) newValues.sum if (current 0) Some(current) else None }注意最后那个None——当状态不再需要时返回 None框架会把它从状态集合里删掉。这个判断条件要结合业务设计比如最近 24 小时无访问则清理就需要在状态里带一个最后访问时间戳而不是只存计数。另一个思路是用mapWithState它的 API 设计更友好有内置的timeout参数可以自动清理超时状态val mappingFunc (userId: String, value: Option[Int], state: State[Int]) { val sum state.getOption().getOrElse(0) value.getOrElse(0) state.update(sum) (userId, sum) } val stateSpec StateSpec.function(mappingFunc).timeout(Seconds(3600)) val result pairs.mapWithState(stateSpec)mapWithState相比updateStateByKey有三个实测优势支持超时清理、只对发生变化的 key 做处理而不是全量遍历、可以拿到状态变更流。它的性能在状态规模大的时候差距非常明显我做过一次对比3000 万 key 的场景下前者单批耗时是后者的三分之一左右。新写的任务没有理由再用updateStateByKey。4.2 checkpoint 的四个真实坑checkpoint 在 Spark Streaming 里承担两件事一是保存流计算的元信息DAG、批间隔、未处理的批二是保存状态数据。开启方式很朴素ssc.checkpoint(hdfs:///streaming/checkpoint/order-stat-v3) ssc.sparkContext.setCheckpointDir(hdfs:///streaming/checkpoint/order-stat-v3)但真正出问题的地方在细节。以下四条我都实打实撞过第一代码升级不兼容。checkpoint 里序列化的是整个 DAG 和算子闭包。你改了类名、删了字段、换了算子用旧 checkpoint 启动会直接反序列化失败报ClassNotFoundException或者InvalidClassException。而且这个错误发生在启动阶段任务根本起不来。结论每次涉及状态结构的代码变更必须换一个 checkpoint 目录用版本号命名是最省心的做法上面例子里的v3就是这个意思。第二checkpoint 目录和代码版本必须一一对应不能共用。我见过两个任务图省事共用一个目录结果状态互相污染跑出来的数字对不上排查了整整两天。第三checkpoint 的写入频率和状态大小直接挂钩。有状态的任务每批都要写一次 checkpoint如果状态有 10GB每 5 秒写 10GB 到分布式存储这个 IO 压力会非常可观而且会直接计入批处理时间。这种情况下要么拉长批间隔要么改用 RocksDB 之类的本地状态存储要么就得考虑迁移到支持增量 checkpoint 的方案。第四从 checkpoint 恢复的耗时不受控。状态越大恢复越慢。10GB 的状态恢复可能要几分钟这段时间位点是不动的如果上游数据保留期短恢复完可能已经消费不到了。所以上游消息的保留时长必须大于你的最坏恢复时间这个值建议至少留 24 小时。4.3 状态里到底该存什么结构选型比容量更重要状态不是越多越好。我判断一条信息该不该进状态只问三个问题它是不是后续批次必须用到的只是当前批用的不进状态它能不能从别的信息推导出来能推导的只存推导源不存结果它的精度要求是什么累计计数用 Long去重用的集合要慎用 Set。第三条尤其重要。用Set[String]存去重集合是很常见的写法但集合大小没有上限跑久了能把内存吃穿。我一般会换成定长位图或者基数估计结构比如 HyperLogLog误差 1% 以内内存固定几 KB。做 UV 统计的时候这个替换能把状态从几十 GB 压到几百 MB误差在业务能接受的范围内。还有一点状态里的对象要尽量精简。Scala 的 case class 每个字段都有开销字符串尤其占地方。能存 id 就不存整个对象能存数字就不存字符串这些细节在状态上千万 key 的时候会体现出数量级的差异。5. 批处理时间的压缩路径一次从 40 秒到 3 秒的排查调优这件事最忌讳上来就改参数。正确的顺序永远是先定位瓶颈在哪一段时间里再针对性地改。Spark Streaming 的时间消耗可以拆成三段每一段的排查手段完全不同。5.1 先分清三种延迟调度延迟、处理延迟、反压Spark UI 的 Streaming 页面里有两个关键指标Scheduling Delay调度延迟和Processing Time处理时间。判断方式很清晰处理时间稳定小于批间隔调度延迟接近 0 —— 健康处理时间偶尔超批间隔调度延迟小幅波动 —— 偶发抖动多半是 GC 或外部依赖变慢处理时间持续接近或超过批间隔调度延迟单调增长 —— 已经积压必须处理。另外看 batch 的输入数据量曲线。如果处理时间在涨但输入量平稳说明是状态规模或外部依赖的问题如果处理时间和输入量同步涨落说明是计算能力不足得加并行度或者优化算子。反压backpressure是另一个必须开的开关spark.streaming.backpressure.enabledtrue spark.streaming.backpressure.initialRate10000 spark.streaming.kafka.maxRatePerPartition5000反压开启后框架会根据上一批的处理速度动态调整下一批的消费量避免把集群打垮。initialRate是启动时的初始速率必须设否则第一波全量涌入很容易直接把任务打死。maxRatePerPartition是每个 Kafka 分区的消费上限单位是条/秒这是硬限速比反压更刚性两者配合用。我一般按这个方式估算maxRatePerPartition单批处理能力条/秒÷ 分区数 × 0.7。比如压测出单批能处理 60000 条/秒24 个分区那就是 60000 ÷ 24 × 0.7 ≈ 1750。留 30% 余量是给突发流量和 GC 停顿的。5.2 并行度的匹配计算核心数、分区数、task 数的三角关系Direct 模式下有个硬约束经常被忽略每个 Kafka 分区在一批内只对应一个 task。所以如果你的 Kafka 有 24 个分区集群给了 96 个核心那这一批最多只能跑 24 个 task剩下 72 个核全程闲着。这不是 Spark 的问题是数据源分区数决定的。规律很直接期望并行 task 数 ≈ Kafka 分区数 ×1 到 3 Kafka 分区数应 ≥ 集群总核数为什么建议 2 到 3 倍因为 task 执行时间总有不均衡分区数刚好等于核数时任何一个小抖动都会让部分核空转。多几个分区能让调度器有腾挪空间。但也不能太多分区过多会让每批的 task 调度开销占比上升小批量场景下反而变慢。计算公式可以这样推设集群总核数 C批间隔 T 秒期望单 task 处理的数据量约 D 条峰值 QPS 为 Q则每批总量 Q × T 分区数 P ≈ 每批总量 ÷ D 且 P ≥ C × 2举例Q 5 万条/秒T 5 秒每批 25 万条。若单个 task 舒适处理量是 8000 条则 P ≈ 31。集群总核数假设 2424 × 2 48 31所以按 48 个分区取整到 48 或 50。最终 Kafka 分区数取 48同时调大 Executor 数量让总核数接近 48 到 72 之间。这里有个实操细节改 Kafka 分区数会影响 key 的路由。如果上游是按 key 分区的扩分区会导致同一 key 的历史数据和新数据落在不同分区做有状态聚合时状态会错乱。所以扩分区前必须确认下游是不是有状态计算有的话要选在状态可清空的时点做或者干脆用新的 topic 灌一遍。5.3 内存与序列化最容易被忽视的 30%Kryo 序列化几乎是 Spark 调优里的免费午餐但很多任务还在用默认的 Java 序列化。spark.serializerorg.apache.spark.serializer.KryoSerializer spark.kryo.registrationRequiredfalse spark.kryo.referenceTrackingfalseregistrationRequiredfalse让 Kryo 自动处理未注册的类省去手工注册的麻烦代价是稍微多一点开销。referenceTrackingfalse关掉引用跟踪对没有循环引用的数据结构能再节省一部分空间和时间。实际测下来这两项配上网络传输和状态序列化的开销通常能降 20% 到 40%。内存配置上批处理任务和执行任务的诉求不太一样。DStream 的接收数据Direct 模式其实不怎么占加上 shuffle 和聚合的中间结果构成了主要占用。我的经验配比是spark.executor.memory8g spark.executor.memoryFraction0.6 spark.memory.storageFraction0.4 spark.executor.cores4 spark.executor.instances12memoryFraction0.6是给执行shuffle、join、聚合留的比例比默认的 0.6 差不多但如果任务全是窗口聚合而几乎没有 shuffle可以调低到 0.4 把空间让给存储。反过来如果全是宽依赖的 join可以到 0.7。GC 是最隐蔽的杀手。判断方法很简单在 Spark UI 的 Executor 页面看 GC Time 占比超过 10% 就得处理。常见原因有三个——频繁创建大对象、状态结构太臃肿、堆外内存没用上。前两个靠改代码第三个可以开spark.memory.offHeap.enabledtrue spark.memory.offHeap.size2g堆外内存能绕开 JVM 的 GC对那种状态大、对象多的任务效果明显。但要注意开了堆外之后spark.executor.memory的核算方式会变容器内存限制得同步调大否则会被系统 OOM Killer 干掉表现为 Executor 无缘无故消失日志里什么都看不到。5.4 一次真实的排查链路说一个我印象最深的案例。任务订单状态实时聚合批间隔 5 秒Kafka 48 分区Executor 12 × 4 核。上线一周后处理时间从 3 秒涨到 40 秒队列积压到几十批。排查顺序是这样的第一步看是不是数据量涨了。拉出输入速率曲线很平稳排除。第二步看状态规模。这是个有状态任务用mapWithState统计每个订单的累计状态。查了一下状态里的 key 数量——4800 万而且只有进没有出。找到第一个原因没有设置超时清理已完结的订单状态永远留在那里。加上 24 小时超时后状态降到 300 万处理时间从 40 秒降到 18 秒。第三步看 GC。降到 18 秒之后 GC 占比还有 25%。dump 了一份堆发现状态 value 里存了一个嵌套的 Map每个订单存了完整的操作流水。改成只存聚合后的几个数值字段处理时间降到 8 秒GC 降到 6%。第四步看 shuffle。还有 8 秒。用 Spark UI 看 stage发现有一个groupByKey之后接mapValues(_.toList)把一个 key 下的所有记录全拉到一台机器上。改成reduceByKey先在 map 端聚合处理时间降到 4 秒。第五步看外部调用。还剩 4 秒。foreachRDD里有个维表查询走的是同步 HTTP 调用每个分区串行执行。改成批量查询加本地缓存并设置了 200ms 超时降级处理时间降到 3 秒。从 40 秒到 3 秒五个原因叠加一个参数都没调全是代码层面的问题。这个案例说明一件事大部分流任务的性能问题不在 Spark 的参数里在你的代码逻辑里。参数调优是最后 10% 的事前面 90% 得从状态管理、算子选择、外部依赖这三个方向找。6. 故障恢复与一致性at-least-once 之后还差什么流任务的稳定性不能靠不出故障来保证只能靠出了故障能正确恢复。这一节聊恢复机制和一致性语义因为它决定了你的数据能不能对得上账。6.1 从 checkpoint 恢复时到底发生了什么任务挂掉后重启如果指定了 checkpoint 目录Spark Streaming 会从最近一次成功的 checkpoint 恢复。恢复的内容包括流计算的 DAG 结构、批间隔、尚未处理的批的元信息、以及状态数据。位点方面Direct 模式下位点存在 checkpoint 里恢复后会从那批开始重新消费。这个过程有两个必须知道的限制。第一checkpoint 是每批写一次的但写入是异步的最后几批可能没写进去。所以恢复后通常会重放最近几批这就是重复的来源之一。第二恢复时的那一批数据必须仍然在 Kafka 里。如果上游保留期是 1 小时而恢复花了 2 小时那批数据已经被清理任务会从auto.offset.reset指定的位置重新开始要么丢数据latest要么从头再来earliest。所以有个硬性运维要求消息保留时长 最坏恢复时长 × 2。最坏恢复时长包括发现故障、拉人、定位、重启、状态加载。这个链路的实际时间往往比想象中长得多我一般按 24 小时保留期起步。6.2 端到端精确一次需要同时满足三个条件Spark Streaming 官方文档会强调可以做到 exactly-once但那是有前提的。严格来说要同时满足条件一数据源可重放。Kafka 天然满足位点可以回退同一段数据可以重新消费。socket 源不满足读走就没了。条件二状态存储可恢复。checkpoint 里保存的状态在恢复后必须一致不能出现位点回退了但状态没回退的情况。这就是为什么改了状态结构必须换 checkpoint 目录——否则语义就断了。条件三输出端幂等或支持事务。这是最容易被忽略的一条。即使前面两条都满足如果输出是INSERT到一张没有唯一键的表重放一次就多一批数据实际效果还是 at-least-once。输出方式一致性等级实现要点无唯一键的 INSERTat-least-once一定重复只适合日志类有唯一键的 upsert幂等等效精确一次唯一键设计要到位事务 位点同写精确一次位点与数据同一事务提交两阶段提交外部支持精确一次需要存储端配合复杂度高实践中我绝大多数场景走的是第二行设计好唯一键用 upsert位点存外部存储。这样做不到理论上的精确一次但效果等价实现成本低运维简单。只有在金融对账这类零容忍的场景才会去碰两阶段提交因为它对存储端的要求高链路一出问题排查难度成倍上升。6.3 故障演练该怎么做才有效不做演练的容灾方案等于没有。我的做法是搭一个和生产同构的小集群定期做三类演练第一类杀 Executor。手动 kill 掉一个 Executor观察任务是否自动恢复、有没有重复数据、恢复耗时多少。这个演练暴露最多的问题是内存配置不合理——恢复时集中处理积压批次内存峰值比正常运行时高得多。第二类杀 Driver。直接停掉 Driver 进程验证 checkpoint 恢复是否可用。这个演练会暴露 checkpoint 目录权限、状态序列化兼容性、位点记录是否完整这几类问题。每次代码变更后必须做一遍因为反序列化失败往往只有真恢复一次才能发现。第三类上游断流。停掉 Kafka 生产端观察任务空转时的行为。有些任务在无数据时会出现奇怪的连接超时或者空指针尤其是外部依赖部分没有做空判断的时候。演练里还有个必查项恢复后第一件事是核对数据总量。拿恢复时间段内输入的总条数和输出的总条数对比差异必须在重复容差范围内。这个核对脚本我从第一个流任务开始就写后来成了团队的标准动作。7. DStream 还是 Structured Streaming现在该怎么选这个问题我在过去两年被问得最多。答案不是简单的用新的得看任务的实际形态。7.1 两代 API 的能力边界对比维度DStreamStructured Streaming编程模型RDD 之上的离散流抽象DataFrame / SQL持续增长的表状态管理手动需 checkpoint内置支持 watermark 与超时事件时间处理弱窗口基于处理时间强watermark 处理乱序精确一次需自行保证输出幂等支持端到端精确一次延迟下限一个批间隔秒级微批模式同左连续模式可到毫秒生态融合与 Dataset 混用割裂与批处理、机器学习统一运维复杂度中checkpoint 坑多中低状态后端可选择排查工具Streaming UI 成熟UI 信息相对少一些从表里能看出Structured Streaming 在状态管理、事件时间、一致性语义这三块是全面的代际优势。尤其是 watermark——DStream 时代处理乱序数据要自己写一堆逻辑判断现在一句withWatermark就能搞定。7.2 什么情况下值得迁移什么情况下别折腾值得迁的信号很明显现在为了处理乱序或迟到数据代码里塞满了自定义逻辑状态规模在涨updateStateByKey已经撑不住需要和批处理、机器学习共用一套代码需要精确一次语义现有的输出幂等方案维护成本高。不值得迁的信号也很明确任务逻辑简单就是一个 map 加一个 reduce跑得很稳强依赖某些 RDD 级别的底层操作比如需要拿到分区信息做自定义处理周边系统监控、调度、告警都围绕 DStream 建好了迁移带来的收益抵不过改造成本团队成员对 DataFrame API 不熟迁移风险大于收益。我实际的判断标准是这个任务半年内还会不会大改会大改趁早迁到 Structured Streaming新功能直接在那边写不会大改就先让它跑着等下次有重构窗口再说。硬迁一堆稳定运行的老任务纯粹是给自己找事。迁移的实操路径建议是双写对比新老任务同时跑输出到不同的目标表用同一份源数据对账一周数字完全对上了再切流量。这个过程中最容易出问题的地方是时间语义——老代码里的窗口如果基于处理时间新代码切到事件时间后结果会不一样必须在迁移前和业务方把口径确认清楚。7.3 迁移时最容易踩的两个坑第一个是foreachBatch里的重复调用。Structured Streaming 的foreachBatch在一个批次重试时会被再次调用如果你的实现里有非幂等的操作比如自增计数器、追加写重试就会产生重复。解决办法是在foreachBatch内部做幂等或者用batchId做去重判断。第二个是watermark 与状态清理的关系。Structured Streaming 的状态清理依赖 watermark 推进而 watermark 又依赖数据里的事件时间。如果上游数据的事件时间整体偏晚或者有大量无界延迟的数据watermark 会卡住不推进状态就会无限膨胀。这个问题的表现和老框架的updateStateByKey不清理一模一样只是原因不同。上线前必须用真实数据分布验证 watermark 的推进速度。写到这里我最大的一个体会是流处理的难点从来不在怎么把数据读进来、算完、写出去那条链路几十行代码就通了。真正花时间的是——批间隔定多少、状态什么时候清、位点由谁管、重复了怎么办、挂掉之后多久能回来。这些问题没有标准答案只能靠对业务的判断和对框架行为的熟悉程度一点点磨出来。我现在的习惯是每接一个新流任务先把最坏情况想一遍如果数据延迟半小时到达会怎样、如果任务连续挂了三次位点会退到哪、如果上游改了大半夜会发生什么。把这些想清楚再动手写代码比事后排查省的时间多得多。