企业官网建设流程全解析

第一次接手实时链路时，团队的离线任务已经堆成山，业务方的诉求从“昨天的报表今天早上八点前必须出来”变成“刚才那一小时的订单异常我现在就要知道”。离线那套跑得再稳，也架不住这种节奏，于是 Spark Streaming 成了当时的默认选项——集群、调度、监控、运维体系全是 Spark 那一套，不用再引进一套全新组件，学习成本也能压在团队内部消化。这篇文章就把这些年用 Spark Streaming 做流式处理的完整思路拆开讲：微批模型到底怎么运转、DStream 这套抽象该怎么用、对接 Kafka 时 offset 怎么管、批次堆积了从哪查起、资源参数怎么算，以及什么情况下应该果断换 Structured Streaming。刚接触流处理的同学可以顺着看，已经在线上跑任务的同学可以直接跳到调优和排查那两节，那里基本是我踩过的坑的合集。

1. 先想清楚：Spark Streaming 到底解决什么问题

1.1 微批模型：把无界的流切成一块一块的批

Spark Streaming 的核心设计思路非常朴素：它不追求逐条事件处理，而是按固定时间间隔把无界的数据流切成一个个小批次（batch），每个批次在内部被包装成一个 RDD，然后交给 Spark Core 的调度器去执行。这个时间间隔就是batchDuration，通常设为 1 秒、5 秒或者 10 秒。理解这一点非常关键，因为后面所有的性能问题、延迟问题、状态问题，本质都源于这个“切”的动作。

打个生活化的比方：真正的流处理像是水龙头一直开着，水流不断；而 Spark Streaming 像在水龙头下面放一个带刻度的桶，每 5 秒把桶里的水倒出来称一次重量。它对“每一滴水什么时候落下来”不关心，只关心“这 5 秒内一共接了多少水、这些水有什么特征”。所以它天然适合做聚合类、统计类、窗口类的计算，比如“每 5 分钟统计一次各地区的下单量”“过去一小时某个接口的错误率”。反过来，如果你需要的是毫秒级的单事件响应，比如风控里的实时拦截、交易撮合，那微批模型的调度开销就会成为天花板，因为即便是最短的批次间隔，从数据进入、任务调度、序列化、执行到结果输出，也至少要几十到几百毫秒，而且这个延迟是抖动的。

微批模型的另一个隐含特性是：一个批次内的数据是有明确边界的。这个边界既是优势也是限制。优势在于，批内可以做非常高效的向量化计算、可以复用 Spark Core 那一整套成熟的 shuffle、join、聚合算子，代码写起来跟批处理几乎一样，团队里会写 Spark SQL 的人上手成本极低。限制在于，跨批次的状态维护必须显式声明，你必须告诉框架“我要保留上一批的状态”，否则每个批次都是独立的。而且批次是有序执行的，前一批没跑完，后一批只能排队等，这就是后面要讲的“批次堆积”问题的根源。

再说说延迟。很多人对 Spark Streaming 的延迟有误解，以为设了 1 秒的批次间隔就是 1 秒延迟。实际端到端延迟大致等于batchDuration加上单批次处理耗时，再加上上下游的传输开销。如果单批次处理耗时是 3 秒，而你设的批次间隔是 1 秒，那么每批的实际间隔就变成了 4 秒左右，延迟会持续累积，直到触发反压机制把消费速率降下来，形成一个动态平衡。所以调优的核心不是把batchDuration设得越小越好，而是让“单批次处理耗时”稳稳地低于批次间隔，留出 30% 到 50% 的余量。

1.2 选型考量：什么时候用它，什么时候果断换

选型这件事，我的判断标准从来不是“哪个技术先进”，而是“团队现有能力栈是什么、业务能容忍多长的延迟、数据量级在哪个区间”。Spark Streaming 的甜点区非常清晰：数据量在每秒几万到几十万条、端到端延迟要求在秒级到十秒级、计算逻辑偏复杂（涉及多流 join、状态聚合、维度关联）、团队已经有 Spark 批处理经验。在这个区间里，它的性价比几乎没有对手，因为你可以复用同一套集群、同一套 UDF、同一套监控。

但如果业务要求延迟压到 100 毫秒以内，或者数据量级到了每秒百万条以上并且需要极低的调度开销，Spark Streaming 就不合适了。前者是因为微批的调度开销摆在那，后者是因为每个批次的 task 数会爆炸，调度器本身会成为瓶颈。还有一种情况是团队没有 Spark 基础，纯新项目，那我会直接推荐 Structured Streaming，因为它是 Spark 官方主推的方向，API 更简洁，事件时间语义和 watermark 都是一等公民，而 DStream 这套 API 在近几个大版本里基本处于维护状态，功能不再演进。

这里要强调一点：很多人把“至少一次”当成缺陷，其实在绝大多数业务里，配合下游的幂等写入（比如按业务主键做 upsert、用唯一键去重），效果跟精确一次没差别。真正不能容忍重复的场景，比如计费扣款，那你需要的是端到端的幂等设计，而不是指望框架给你兜底。这个认知我花了很久才建立起来，早期一度执着于“框架必须保证精确一次”，结果在业务侧反而没做去重，出过重复计费的问题。

2. 核心抽象拆解：DStream、批次与执行链路

2.1 DStream 的本质是 RDD 序列

DStream（Discretized Stream）这个名字听着抽象，实际上它的定义简单得让人意外：它就是一个按时间顺序排列的 RDD 序列。你在代码里写的每一个map、filter、reduceByKey，框架都会记录下来，形成一个 DAG 模板，然后在每个批次触发时，把这个模板实例化到当前批次的 RDD 上执行。所以你在 DStream 上调用的算子，本质上是在定义“每个批次该做什么”，而不是“对这条数据做什么”。

这个认知决定了你写代码的思维方式。举个最容易出错的例子：很多人会在foreachRDD外面创建一个数据库连接，然后在foreachRDD里面用。这段代码在 driver 端能跑，但一到 executor 端就报序列化异常，因为连接对象是不可序列化的，而foreachRDD里的闭包要被分发到 executor 执行。正确的做法是在rdd.foreachPartition里面创建连接，让每个 partition 复用一个连接，既避免序列化问题，又避免每条记录建一次连接的巨大开销。

另一个容易被忽视的点是 DStream 的惰性。定义 DStream 的时候什么都不会发生，只有调用了ssc.start()之后，框架才会按批次间隔不断触发。这也意味着，你在start()之前打印一个 DStream，是看不到任何数据的，只能看到类型信息。调试的时候我习惯用print()算子，它会打印每个批次前 10 条数据，虽然简陋但非常直观。

DStream 的血缘关系是逐批次建立的，每个批次的 RDD 都独立维护自己的 lineage。这意味着如果某个批次的某个 partition 计算失败，Spark 可以只重算那一个 partition，而不需要重跑整个批次。但同时也要注意，如果 lineage 拉得太长（比如连续几十个算子），重算的成本会很高，适当用persist截断血缘是有必要的，尤其是在有状态操作的场景下。

2.2 一个批次从进入到输出的完整链路

搞清楚一个批次的生命周期，排查问题时就有了地图。以最常见的 Kafka 直连模式为例，一个批次的旅程大致是这样：Receiver 或者 Direct API 在批次触发时向 Kafka 拉取该批次对应的 offset 区间数据，形成一组 RDD partition；然后框架把你在 DStream 上定义的算子链应用上去，生成 Job；JobScheduler 把 Job 提交给 SparkContext 的 DAGScheduler；DAGScheduler 按 shuffle 边界把 Job 切成若干 Stage，每个 Stage 生成一批 Task 分发到 executor；executor 执行完把结果写出去或者回传给 driver；driver 收到所有 Task 完成后，记录这个批次完成，然后触发 offset 提交和 checkpoint 写入。

这条链路上任何一个环节变慢，都会体现为批次处理时间变长。所以排查问题的第一步永远是看 Spark UI 的 Streaming 页面，那里有每个批次的处理时间、调度延迟、输入速率这几条曲线。如果“处理时间”曲线持续逼近甚至超过批次间隔，那问题在计算侧；如果“调度延迟”很高但处理时间不高，那问题在资源不足或者前面有别的任务在抢资源；如果输入速率突然飙升而处理时间跟着涨，那是流量突增，该考虑反压或者扩容。

我见过最多的误判是把“调度延迟高”当成“计算慢”，然后拼命优化算子逻辑，结果毫无效果。实际上调度延迟高往往是因为批次内的 task 数超过了集群能同时提供的槽位，后面的 task 只能排队等。这时候的解法是加 executor 或者减少 task 数（比如降低 Kafka 分区数、提高合并度），而不是优化代码。区分这两个指标，能省掉你一半的排查时间。

还有一点值得注意：批次是串行执行的。框架会等当前批次的所有 Job 都完成后，才会触发下一个批次。这个设计保证了批次间的顺序性，但也意味着没有天然的并行度补偿。如果某一批因为数据倾斜跑了 30 秒，那后面 30 个批次（假设 1 秒间隔）都会被阻塞，然后一次性涌进来，形成雪崩。这就是为什么反压机制和合理的批次间隔设置如此重要。

2.3 状态与窗口：把跨批次的信息留下来

无状态算子处理的是“当前批次内”的数据，而有状态算子要跨批次保留信息，典型场景是“统计每个用户从今天零点到现在的累计下单金额”。Spark Streaming 提供了两代方案：老一些的updateStateByKey和新一些的mapWithState。前者每次都要对全量 key 做一次 shuffle，数据量一大就撑不住；后者引入了State对象和超时机制，只处理有变化的 key，性能和内存表现都好得多，我在生产环境里基本只用mapWithState。

窗口操作是另一块常用能力。window(windowLength, slideInterval)定义一个长度和滑动步长，reduceByKeyAndWindow在窗口上做聚合。这里有个非常重要的优化点：如果你能把窗口的聚合拆成“加新数据”和“减旧数据”两个操作（也就是提供invFunc反向函数），框架就只需要处理进出的增量，而不需要每次重算整个窗口。这个优化在长窗口场景下能把计算量降低一到两个数量级，比如“过去 24 小时的累计值，每分钟更新一次”，不优化的话每一分钟都要重算 24 小时的数据。

有状态操作必须开启 checkpoint。这不是可选项，是硬性要求。因为状态数据本身要持久化，driver 挂了之后要能从 checkpoint 恢复状态和 DStream 的图结构。但 checkpoint 也是个大坑，后面第 4 节会专门讲。这里先埋个提醒：checkpoint 目录一旦启用，就绑定了你当时的代码逻辑，改了算子链之后恢复会失败，必须清空目录重来。

3. 实操：从零搭一条 Kafka 到 Spark Streaming 的准实时链路

3.1 环境与依赖版本对齐

版本对齐这件事，看起来是琐碎的准备工作，实际上是新手最容易栽跟头的地方。Spark Streaming 对接 Kafka 依赖spark-streaming-kafka-0-10这个包，它内部又依赖kafka-clients的特定版本。如果你的 Spark 版本是 2.4.x，配套的 kafka-clients 大致在 2.0 左右；如果是 Spark 3.x，kafka-clients 会到 2.x 的更高版本。版本错配最典型的表现是运行时报NoSuchMethodError或者ClassNotFoundException，而且报错信息往往指向一个你根本没直接调用的类，非常迷惑。

我常用的做法是先用spark-submit --version确认 Spark 版本，再去看对应的官方依赖矩阵，然后在构建文件里显式锁定 kafka-clients 的版本，避免传递依赖把版本拉乱。如果是 Maven 项目，用dependency:tree检查一下有没有多个 kafka-clients 版本共存；如果是 SBT，evicted命令能直接列出冲突。这一步花十分钟，能省掉后面半天的调试。

<!-- 以 Spark 3.x 为例，版本需与实际集群对齐 --> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming_2.12</artifactId> <version>3.3.2</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming-kafka-0-10_2.12</artifactId> <version>3.3.2</version> </dependency> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.8.1</version> </dependency>

注意spark-streaming那个依赖要标provided，因为集群上已经有这个包了，打进 fat jar 反而可能和集群版本冲突。而 kafka 相关的包集群上没有，必须打进去。这个区分很基础，但我见过不止一个团队把所有依赖都打成 fat jar，结果运行时加载了旧版本的 Spark 类，出现各种诡异行为。

3.2 直接消费 Kafka 的代码骨架

消费模式有两种：基于 Receiver 的老模式，和基于 Direct API 的新模式。前者用 ZooKeeper 管理 offset，依赖 WAL 保证不丢数据，配置复杂且性能一般；后者直接查询 Kafka 的分区和 offset，自己管理消费位置，语义更清晰，性能也更好。新项目一律用 Direct 模式，老项目如果还在用 Receiver，建议尽早迁移。

下面是 Direct 模式的骨架，核心是LocationStrategies和ConsumerStrategies这两个策略对象。

val conf = new SparkConf() .setAppName("order-realtime-stat") .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") .set("spark.streaming.backpressure.enabled", "true") .set("spark.streaming.kafka.maxRatePerPartition", "2000") .set("spark.streaming.stopGracefullyOnShutdown", "true") val ssc = new StreamingContext(conf, Seconds(5)) ssc.checkpoint("hdfs:///checkpoint/order-realtime-stat") val kafkaParams = Map[String, Object]( "bootstrap.servers" -> "broker-a:9092,broker-b:9092", "key.deserializer" -> classOf[StringDeserializer], "value.deserializer" -> classOf[StringDeserializer], "group.id" -> "order-realtime-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) )

LocationStrategies.PreferConsistent的意思是尽量把分区均匀分配到所有 executor 上，这是默认也是大多数场景下最合适的选择。如果你发现某些 executor 上的消费特别慢，可以换成PreferFixed手动指定，但一般不建议，因为手动分配很容易造成负载不均。enable.auto.commit必须关掉，让框架和你自己的代码来控制提交时机，否则会出现“数据处理失败了但 offset 已经提交”的丢数据问题。

拿到stream之后，第一步通常是提取 offset 和业务数据。stream.foreachRDD里可以拿到rdd.asInstanceOf[HasOffsetRanges].offsetRanges，这是整条链路上唯一能准确拿到 offset 信息的位置。一定要在第一个foreachRDD里就把 offset 提取出来，因为只要经过任何map或filter，RDD 的类型就变了，HasOffsetRanges的转换会失败。

3.3 offset 管理与断点续跑

offset 管理是流处理里最考验设计功力的部分。核心原则只有一条：offset 的提交必须发生在数据处理成功之后。听起来简单，但“成功”的定义在不同场景下差别很大。如果你的下游是幂等的（比如按主键 upsert 到数据库），那至少一次就够了，处理完就提交；如果下游不幂等，那要么做事务性写入，要么在提交前做好去重。

存储位置的选择也有讲究。常见的有三种：Kafka 自身的__consumer_offsets、ZooKeeper、外部存储（MySQL、HBase、Redis）。我个人的偏好是外部存储，理由是可观测性好——你能直接查表看到每个分区消费到哪了，出问题的时候一目了然。用 Kafka 自身存储的话，查起来麻烦，而且一旦 group 被误删，offset 就没了。用外部存储的代价是要自己处理并发写入，不过量级不大，一般不是瓶颈。

stream.foreachRDD { rdd => val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges // 业务处理：注意所有操作都要在 rdd 内部完成 val result = rdd .map(_.value()) .flatMap(parseOrder) // 解析失败返回 None .filter(_.amount > 0) .map(o => (o.region, o.amount)) .reduceByKey(_ + _) result.foreachPartition { iter => val conn = DriverManager.getConnection(jdbcUrl, user, pwd) try { // 幂等写入：按 (region, window_start) 做 upsert val ps = conn.prepareStatement( "INSERT INTO region_stat(region, stat_time, amount) VALUES(?,?,?) " + "ON DUPLICATE KEY UPDATE amount = VALUES(amount)") iter.foreach { case (region, amount) => ps.setString(1, region) ps.setTimestamp(2, new Timestamp(batchTime.milliseconds)) ps.setBigDecimal(3, amount) ps.addBatch() } ps.executeBatch() } finally conn.close() } // 业务写入成功后再提交 offset storeOffsetToDB(offsetRanges) }

这段代码里有两个细节值得说。第一，foreachPartition里面创建连接，每个 partition 一个连接，比每条记录一个连接高效得多，也不会出现序列化问题。第二，ON DUPLICATE KEY UPDATE这种幂等写入方式是配合至少一次语义的标准做法，重复消费同一批数据时，结果会被覆盖成相同的值，不会产生重复累加。如果你的聚合是累加型的，那就要在表里存一个批次标识，先查再更新，或者用更严格的事务。

还有一个坑是重启时的起始位置。如果你从外部存储读 offset，要注意首次启动（表里没数据）的情况，此时应该走auto.offset.reset的逻辑。另外强烈建议实现一个“从指定时间点回溯”的入口，线上出问题需要补数据的时候，这个功能能救命。我习惯在启动参数里留一个--start-time选项，传了就从那个时间戳对应的 offset 开始消费。

3.4 反压与批次间隔调参

反压（backpressure）是 Spark Streaming 从 1.5 版本开始引入的机制，默认是关闭的，必须显式打开。它的工作原理是根据前几个批次的处理延迟动态估算下一批应该拉多少数据，通过RateEstimator（默认是 PID 控制器）计算出一个速率上限，然后限制每个 Kafka 分区每秒钟最多拉多少条。

打开反压只需要两个配置：spark.streaming.backpressure.enabled=true，以及可选的spark.streaming.backpressure.initialRate用来设置首次启动的初始速率。如果不设初始速率，第一批会按maxRatePerPartition全速拉，容易在冷启动时把内存打爆。

spark.streaming.kafka.maxRatePerPartition这个参数是硬上限，单位是“每分区每秒的记录数”。它的作用是给反压兜底，防止在流量突增时反压还没反应过来就把 executor 撑爆。这个值的估算方法是：先测出单分区单批次能稳定处理的最大记录数，除以批次间隔秒数，再打个七折。比如单分区一批（5 秒）能处理 1 万条，那就是 2000 条/秒，七折后设 1400 左右。

批次间隔的调整则要看业务延迟容忍度。有个经验值：如果你发现把批次间隔从 5 秒降到 1 秒之后，吞吐量明显下降，说明调度开销占比太高了，可能每个批次的 task 数太少，CPU 利用率上不去。反过来如果批次间隔调大了但延迟没改善，说明瓶颈不在批次间隔上，在别的地方。

注意：反压不是万能的。它只能调节消费速率，没法解决数据倾斜。如果某个 Kafka 分区因为 key 分布不均导致单 task 处理特别慢，反压会让所有分区一起降速，整体吞吐被拖累。这种情况要先解决分区均衡问题。

3.5 提交脚本与资源参数计算

资源计算这块，我见过太多人凭感觉填数字，结果要么资源浪费一半，要么 task 排队排到天荒地老。给一个具体的计算过程：假设 Kafka topic 有 60 个分区，批次间隔 5 秒，每个分区在 5 秒内平均产生 5000 条记录，那么单批次总量是 30 万条。处理逻辑里有一次 shuffle 聚合，单条记录处理耗时约 0.02 毫秒，单批次纯计算耗时约 6 秒——这就已经超了批次间隔，必须优化或者加资源。

假设优化后单批次耗时压到 2 秒，那么需要的核数怎么算？并发 task 数的上限大致等于“executor 数 × 每 executor 核数”。由于 Kafka 直连模式下每个分区的数据对应一个 task，单批次 task 数至少是 60。如果每个 task 处理耗时 0.5 秒，那理论上 2 个 wave 能跑完，也就是 120 个核。但实际要考虑 shuffle 阶段的额外 task 和资源碎片，一般按 1.5 倍冗余配置，也就是 180 核左右。按每 executor 4 核算，需要 45 个 executor，再留 1 到 2 个给 driver。

spark-submit \ --master yarn \ --deploy-mode cluster \ --class com.example.OrderRealtimeStat \ --driver-memory 2g \ --driver-cores 1 \ --executor-memory 8g \ --executor-cores 4 \ --num-executors 45 \ --conf spark.dynamicAllocation.enabled=false \ --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \ --conf spark.streaming.backpressure.enabled=true \ --conf spark.streaming.kafka.maxRatePerPartition=1400 \ --conf spark.streaming.stopGracefullyOnShutdown=true \ --conf spark.sql.shuffle.partitions=120 \ order-stat.jar --start-time "2024-05-01 00:00:00"

这里有两点值得说明。第一，流任务一般要关掉动态资源分配。因为流任务是持续运行的，动态分配在缩容时可能会把正在处理数据的 executor 回收掉，导致批次超时甚至任务失败。第二，spark.sql.shuffle.partitions要单独调，默认 200 对于流任务来说往往太多，每个批次生成 200 个 shuffle task 但每个 task 只有很少的数据，调度开销会吃掉大部分时间。设成 executor 核数的一到两倍比较合适。

4. 常见问题与排查技巧实录

4.1 批次堆积与延迟飙高的排查路径

批次堆积是最常见的线上问题，表现形式是 Spark UI 上“处理时间”曲线持续高于批次间隔，输入队列里排队的批次越来越多。排查我一般按这个顺序走。

第一步看是不是流量真的涨了。对比“输入速率”曲线和处理时间曲线的拐点，如果两者同步上涨，那就是流量问题，短期靠反压降速、长期靠扩容解决。第二步看是不是数据倾斜。打开某个慢批次的 Stage 详情，看 task 的执行时间分布，如果最大值和中间值差了三倍以上，基本可以确定有倾斜，去看 shuffle read size 最大的那几个 task 对应的 key。第三步看 GC。如果 executor 的 GC 时间占比超过 10%，说明内存压力大，可能是缓存了太多数据或者对象创建太频繁。第四步看外部依赖。下游数据库慢、网络抖动、Kafka broker 负载高，都会体现为处理时间变长，这时候要去看下游的监控，而不是死盯着 Spark。

有一个特别隐蔽的情况：状态无限增长。mapWithState如果不设置超时，用户维度的状态会一直累积，跑了几天之后状态数据涨到几个 G，每批次的 checkpoint 写入时间越来越长，最后拖垮整个任务。我现在的标准做法是给所有有状态操作都设置一个合理的超时，比如“30 分钟无更新就清理”，并且定期观察 checkpoint 目录的大小趋势。

4.2 数据重复与丢失的典型坑

数据丢失的经典场景是 offset 提前提交。比如在foreachRDD里先commitAsync再做业务处理，中间如果抛异常，数据就丢了。还有一种更隐蔽的：开了enable.auto.commit=true，Kafka 客户端会按固定间隔自动提交，跟你业务处理完没完完全无关。这个问题在测试环境很难发现，因为测试环境不出错，一到生产环境偶发下游超时，就会丢一批数据。

数据重复的场景就更多了。最常见的三种：一是任务重启时从上次提交的 offset 重新消费，导致最后一批重复；二是foreachRDD里业务处理成功但提交 offset 失败，重启后重放；三是开启了推测执行（speculative execution），同一个 task 被两个 executor 同时跑，结果都写了一遍。第三种在流任务里尤其危险，我现在的做法是流任务一律关掉推测执行，spark.speculation=false。

还有一种重复来源是 checkpoint 恢复。从 checkpoint 恢复时，框架会从 checkpoint 里记录的 offset 重新开始，而这个 offset 可能比外部存储里记录的更旧，导致重放一段数据。这就是为什么我一直强调下游必须做幂等，框架层面很难做到严格的精确一次。

提示：如果你的业务绝对不能容忍重复，最稳妥的方案是在写入侧做唯一约束，而不是在消费侧反复折腾。数据库的唯一索引、Redis 的 SETNX、或者带批次 ID 的去重表，都是成本很低但效果确定的手段。

4.3 checkpoint 相关的坑

checkpoint 目录一旦启用，就会序列化整个 DStreamGraph，包括你写的所有匿名函数。这意味着如果你改了代码逻辑然后从旧 checkpoint 恢复，大概率会报ClassNotFoundException或者反序列化失败，更麻烦的是有时候不报错但行为异常。这个设计在开发阶段特别折磨人，因为每次改代码都得先删 checkpoint 目录。

我的处理办法是：开发阶段把 checkpoint 目录设成本地临时路径，方便随时删；生产环境把 checkpoint 路径带版本号，比如/checkpoint/order-stat/v3，每次上线新版本就换一个路径，旧路径保留一段时间用于回滚。同时把“是否从 checkpoint 恢复”做成启动参数，需要重置状态的时候显式指定从头开始，避免误操作。

另一个坑是 checkpoint 的写入频率。每批次都要写一次，如果状态很大，写入本身就会成为瓶颈。这时候可以考虑用mapWithState替代updateStateByKey，因为前者可以只写增量；或者把状态存到外部存储（比如 Redis），checkpoint 只用来恢复 DStreamGraph，不存实际状态数据。

还有一个容易忽略的点：checkpoint 目录必须对所有 executor 可见，通常是 HDFS 路径。如果用了本地路径，driver 挂掉重启后换了一台机器，就找不到 checkpoint 了。这个问题在多租户集群上尤其常见，因为 driver 可能被调度到任意节点。

5. 性能调优与向 Structured Streaming 迁移的取舍

5.1 序列化、并行度与内存这三板斧

调优这件事，说到底就是在序列化、并行度、内存这三者之间找平衡。序列化方面，Kryo 几乎是必选项，比 Java 原生序列化快好几倍，体积也小很多。配置上除了spark.serializer，还要记得spark.kyro.registrationRequired=false，否则没注册的类会退化成 Java 序列化。如果要追求极致，可以把常用的业务类注册进去，能再省一点空间。

并行度方面，核心是让每个批次的 task 数跟集群的核数匹配。task 太少，CPU 用不满；task 太多，调度开销大。对于 Kafka 直连模式，task 数天然等于分区数，所以 Kafka topic 的分区数其实决定了你的并行度上限。如果发现分区数太少导致跑不满集群，可以在消费之后加一步repartition，但这会引入一次额外的 shuffle，要权衡。

内存方面，流任务的 executor 内存大部分被两部分吃掉：缓存的数据和 shuffle 的中间结果。如果开了persist，要注意存储级别，MEMORY_ONLY在内存不足时会丢分区然后重算，流任务里重算的代价很高，所以一般用MEMORY_AND_DISK_SER。另外spark.memory.fraction这个参数在流任务里可以适当调高到 0.7 左右，因为流任务很少有大量用户自定义对象常驻内存。

还有一个容易被忽略的调优点：减少不必要的 shuffle。流任务里每一次reduceByKey、groupByKey、join都是一次 shuffle，每一次 shuffle 都要落盘、走网络。能用mapPartitions局部聚合的先局部聚合，能用 broadcast join 的绝不用 shuffle join，维度表这类小表一定要广播出去。我在一个项目里把一个 shuffle join 改成 broadcast join 之后，单批次处理时间从 4 秒降到 1.2 秒，效果非常直接。

5.2 迁移判断与双跑方案

Spark Streaming 目前处于维护状态，新特性基本都往 Structured Streaming 上走。但这不意味着所有存量任务都要立刻迁移，迁移成本和风险是实打实的。我的判断标准有三条：如果现有任务稳定运行、没有新需求、性能也够用，那就别动，稳定压倒一切；如果现有任务频繁出问题、或者有新需求要加复杂的事件时间处理，那就借着新需求的机会迁移；如果是全新项目，直接上 Structured Streaming，没有再选 DStream 的理由。

迁移的技术路径上，最大的变化是 API 从 DStream 变成 DataFrame/Dataset，状态操作从mapWithState变成flatMapGroupsWithState或者直接用内置的聚合算子，offset 管理从手动变成框架托管。好处是代码量大幅减少，事件时间和 watermark 是原生支持的，exactly-once 的 sink 也有现成实现。坏处是有些在 DStream 里很灵活的写法，在 Structured Streaming 里要用 UDF 或者 foreachBatch 绕一下。

我一般建议的做法是双跑一段时间。新链路先以旁路方式接入，把结果写到另一张表，然后跟老链路的结果做比对，比对通过之后再切换流量。这个过程中要注意两边的批次语义不同，DStream 的批次是处理时间对齐的，Structured Streaming 默认也是处理时间触发，但窗口语义基于事件时间，所以统计口径可能对不上，需要对时间窗口做对齐处理，否则会看到大量“差异”其实是口径问题。

注意：双跑期间两套任务会同时消费 Kafka，如果用的是同一个 group，会互相抢分区。正确做法是新链路用独立的 group，从最早的 offset 或者指定时间点开始消费，避免影响老链路的进度。

最后分享一个我在实际项目中体会最深的点：流任务的稳定性，八成取决于你对“失败”的处理设计，而不是你算子写得多巧妙。任务一定会重启、下游一定会超时、网络一定会抖动、磁盘一定会写满，这些都是常态。与其花时间追求零故障，不如把重启后的恢复逻辑、offset 的幂等提交、下游的幂等写入这几件事做扎实。我见过太多任务，逻辑写得花里胡哨，但一重启就丢数据或者重复写，最后不得不推倒重来。把这几块地基打牢，剩下的调优才有意义。