刚接手一个实时数据看板项目时我翻遍了各种资料发现讲 Spark Streaming 的不少但真正用 Python 写、从开发到部署踩坑全记录的真不多。所以我就把这两年用 PySpark 跑 Spark Streaming 的实战经验整理成这篇长文从核心概念、环境搭法、常用算子到生产调优、问题排查一次性聊透希望能帮到正在用 Python 做实时计算的朋友少走几个弯路。这篇文章适合的人刚接触流式计算想快速上手的开发者、已经会用 Spark 批处理但想迁移到实时场景的工程师、以及正在用 Python 版 Spark Streaming 却频繁踩坑的运维或数据开发。内容不追求概念堆砌而是以可直接参考的核心代码、参数配置和排查思路为主。1. 先搞清楚Spark Streaming 到底在解决什么问题1.1 流处理和批处理的边界以及微批处理的思路很多刚接触实时计算的人会搞混一个概念Spark Streaming 并不是真正逐条处理数据的流引擎它采用的是“微批处理”模式。要理解这一点先看数据形态的区别。传统的批处理拿到的是一份有边界的完整数据集比如昨天的日志文件、上周的订单表跑一轮 MapReduce 或 Spark SQL 出结果。流处理面对的是无边界的连续数据流一条条进来没完没了。Spark Streaming 的思路很直接把无界流切成一个个有界的小批量每隔几秒或十几秒切一刀每个小批量就是一个 RDD然后交给 Spark Core 去跑标准的批处理作业。所以它说到底是“把实时问题转化为连续不断的小批处理问题”。这种设计带来的好处很明显复用 Spark 批处理生态API 和批处理几乎一致容错机制直接继承 RDD 的 Lineage对已经熟悉 Spark 批处理的人来说上手快。坏处也明显延迟不是毫秒级而是取决于你切批的间隔通常秒级起步。如果你的业务需要毫秒级响应比如高频交易风控那得换真正的流引擎但如果是做实时报表、监控告警、用户行为统计这类近实时需求Spark Streaming 的性价比相当高。1.2 DStream 的核心机制以及 Python 版为什么可行在 Spark Streaming 里一切操作都围绕一个叫 DStream 的东西展开。DStream 直译是“离散化数据流”本质是一个连续的 RDD 序列每个时间片生成一个 RDD。你在 DStream 上做的 map、filter、reduceByKey 操作实际上会被翻译成对底层每个 RDD 的批处理操作。理解这一点很重要因为你在代码里写的 transformation最终都要落到 RDD 级别去执行。再说到 Python 版。PySpark 本质是 Python 进程通过 Py4J 调用 JVM 里的 Spark 对象Python 端定义处理逻辑序列化后传到 JVM 端执行。以前很多人担心 Python 跑流作业性能差其实关键在于你写了什么算子。如果主要用 map、filter 这类窄依赖算子序列化开销虽有但还没到不可接受的程度。真正要小心的是那些需要把 Python 函数传进 JVM 做严重计算的地方后面我会专门讲这个坑。Spark Streaming 的运行流程大致是这样的StreamingContext 启动后Receiver 或 Direct API 从数据源拉取数据按设定的批处理间隔生成 DStream 的每个 RDD然后进入 DAG 调度执行最终将结果输出到外部存储。DStream 还提供了 checkpoint 机制定期把状态和元数据写到可靠存储比如 HDFS这样在 Driver 挂掉后可以从中恢复。1.3 适用场景与不合适场景我自己的经验是这几类场景非常适合用 Spark Streaming实时监控与告警比如服务器指标异常检测、接口错误率突增告警。实时指标计算在线用户数、实时成交额、热门商品排名。数据清洗与转发从 Kafka 消费日志做字段过滤、格式统一后写入 Hive 或 MySQL。实时特征计算给下游推荐系统提供实时的用户行为特征。不太适合的场景我也遇到过毫秒级延迟的在线服务链路比如实时风控拦截需要每个请求毫秒内判定Spark Streaming 的秒级延迟做不到。复杂事件处理比如检测一段时间内连续发生的一系列有逻辑关系的事件这类用 CEP 引擎或 Flink 更顺手。需要大量异步 IO 逐条处理的场景Python UDF 加外部接口调用会严重拖慢吞吐。搞清楚这些边界你后面选型就不会跑偏。接下来进入实操环节。2. 环境准备与第一个 Python Streaming 作业2.1 这版环境怎么搭版本兼容性的血泪教训先说大版本选择。如果你是在新项目里从零开始建议直接用 Spark 3.x 搭配 Python 3.8 以上。Spark Streaming 在 Spark 3.x 里已经进入维护模式官方主推的是 Structured Streaming但 Spark StreamingDStream API仍然被完整支持而且很多线上老项目还在大量使用。如果你维护的是老平台那 Spark 2.4 配 Python 3.6 的也不少见。版本对应关系是个大坑我踩过最疼的一次是用 Spark 2.3 跑 Python 3.7结果跑出来的报错毫无头绪最后发现是 Py4J 的 Java 版本不兼容。建议部署前先在测试环境跑通一个最简单的 word count 流作业确认 Python、Java、Spark 三者匹配再往下走。本地开发调试时我习惯用 Spark 的 standalone 本地模式直接spark-submit提交 Python 脚本不需要搭集群。生产环境再用 YARN 或 K8s。依赖方面pyspark 是核心包另外如果你要接 Kafka需要额外引入 kafka 客户端库。Java 和 Python 进程间的通信依赖 Py4J用 spark-submit 时这些会自动配置好。2.2 5 分钟跑通最简单的 Socket 单词计数好直接上个最简单的例子。这个示例从 TCP Socket 读取文本流做词频统计图的是先把整个链路跑通。from pyspark import SparkContext from pyspark.streaming import StreamingContext # 本地模式2个线程 sc SparkContext(local[2], NetworkWordCount) ssc StreamingContext(sc, 5) # 每5秒切一个批次 # 每隔5秒从 localhost:9999 读取数据 lines ssc.socketTextStream(localhost, 9999) # 把每行拆成单词累加计数 words lines.flatMap(lambda line: line.split( )) pairs words.map(lambda word: (word, 1)) word_counts pairs.reduceByKey(lambda a, b: a b) # 打印每批次结果 word_counts.pprint() # 启动流计算 ssc.start() ssc.awaitTermination()运行前先启动一个发送数据的源我一般这样测试nc -lk 9999然后另开终端提交作业spark-submit \ --master local[2] \ network_word_count.py在 nc 窗口随便敲几行英文看控制台输出统计结果。虽然这个例子老掉牙了但它是验证环境通不通最快的方式。注意local[2]里的 2 意味着至少两个线程原因是流作业需要一个线程接收数据另一个线程处理如果只给一个线程作业会一直卡住不输出。这个细节我当初第一次跑的时候没注意调了半天才发现。2.3 项目结构建议跑通之后项目文件别都堆在一个脚本里。我现在的习惯是拆成几层config/放作业参数配置比如 Kafka broker 地址、topic 列表、batch 间隔用 yaml 或 json 维护不写死在代码里。sources/封装数据源初始化逻辑比如不同的 DStream 接入函数。processors/核心的业务转换逻辑每个模块对应一类数据处理链路。sinks/输出逻辑写 MySQL、Redis、Hive 的代码统一封装。main.py组装上面几个模块作为 spark-submit 的入口。我见过不少维护成本高的流作业问题往往出在把所有代码塞进一个文件里改一个字段要全文搜索。实际上 Python 版本的 DStream 代码逻辑和 Java/Scala 版本是等价的你可以像写普通 Python 工程一样去组织代码只要确保模块能被 Spark 的 executor 找到就行。做法上把依赖模块打成 zip 或放到 PYTHONPATH 里spark-submit时用--py-files参数带上。3. 常用数据源接入与核心算子实操3.1 接入 Kafka生产环境最常见的方式Socket 和本地文件只是测试用生产里百分之八十是接 Kafka。Python 版接入 Kafka我推荐用 Direct 方式也就是createDirectStream思路它不再需要 Receiver不再有 WAL 的额外开销而且和 Kafka 分区一一对应天然适合做并行消费。下面给一个从 Kafka 消费的完整例子from pyspark import SparkContext from pyspark.streaming import StreamingContext from pyspark.streaming.kafka import KafkaUtils sc SparkContext(local[2], KafkaStreamExample) ssc StreamingContext(sc, 10) ssc.checkpoint(hdfs://namenode:9000/checkpoint/kafka_demo) # Kafka 参数配置 kafka_params { bootstrap.servers: kafka1:9092,kafka2:9092, group.id: spark-streaming-group, auto.offset.reset: largest, key.deserializer: org.apache.kafka.common.serialization.StringDeserializer, value.deserializer: org.apache.kafka.common.serialization.StringDeserializer } topics {realtime_logs: 1} # 值为分区数用于创建多少个RDD分区 # Direct 方式创建流 kafka_stream KafkaUtils.createDirectStream( ssc, topics, kafka_params, fromOffsetsNone ) # value 是 bytes先转成 str lines kafka_stream.map(lambda x: x[1].decode(utf-8)) lines.pprint() ssc.start() ssc.awaitTermination()注意KafkaUtils.createDirectStream是 Spark 2.x 时代的接口在 Spark 3.x 里如果你还想用 DStream 方式有的发行版已标记废弃或移除需要确认你用的发行版支持情况。因为官方主推 Structured Streaming这部分代码后续迁移平台时是主要改造点。不过我项目上维护的老作业里这种写法仍然跑得很稳这也是我专门把它写出来的原因。如果你是 Spark 3.3 且发行版里没有pyspark.streaming.kafka模块了简单替代方案是直接从 Kafka 用普通消费者把数据拉到本地文件或内存队列再喂给流作业但这种间接方式牺牲了端到端的可靠语义不建议生产使用。也可以考虑直接用 Structured Streaming 的readStream替代如果你没有历史包袱从 Spark 3.x 开始建议直接上这个DStream API 更多的价值在于看懂老代码和迁移需求。3.2 核心转换算子详解map、flatMap、reduceByKey 的流式行为DStream 的算子分两类转换算子和输出算子。转换算子就是返回新的 DStream 的那批方法。map 和 flatMap 是最常用的逐条操作。map 是一进一出flatMap 是一进多出或者零出。比如一行日志拆成多个字段、一段文本拆成多个单词都要用 flatMap。这两个算子处理逻辑简单即便在 Python 端有序列化开销性能也可接受。reduceByKey 做的是同一批次内按键聚合。注意它只在当前微批内聚合不跨批做累计。如果你希望历史数据也参与统计得用窗口算子或状态算子。这个区别非常关键我见过有人用 reduceByKey 实时统计总订单量结果算出来的只是当前批次的数据没有累计效果最终挨个重跑。再看一个实际业务里高频使用的组合先按 key 分组、再做多个维度的聚合。比如同一批数据里你要同时算各地区的订单数和各品类的销售额可以用一次 flatMap 生成多个键比如(order, 北京, 1)和(amount, 食品, 89.5)最后用一个 reduceByKey 统一聚合这样避免对同一份数据跑多个作业。3.3 窗口计算实战实时统计滑动窗口内的指标窗口计算是流处理的核心场景。比如“过去 5 分钟的实时成交额”“最近 10 分钟的活跃用户数”窗口算式就是干这个的。窗口有两个参数窗口长度window length和滑动间隔slide interval。窗口长度决定关联多少历史数据滑动间隔决定每隔多久算一次。按我的理解它俩都得是批处理间隔的整数倍。比如批间隔设 5 秒窗口长度 15 分钟滑动间隔 1 分钟那么每 1 分钟会基于最近 15 分钟的数据计算一次结果这就是典型的滑动窗口。如果窗口长度等于滑动间隔就退化成滚动窗口只在窗口边界出一次结果。下面是一个实时统计每分钟成交额的例子数据源是 Kafka消息格式为timestamp,city,amount的字符串。from pyspark import SparkContext from pyspark.streaming import StreamingContext from pyspark.streaming.kafka import KafkaUtils sc SparkContext(local[2], SalesWindowStat) ssc StreamingContext(sc, 5) lines KafkaUtils.createDirectStream(ssc, {sales: 3}, { bootstrap.servers: kafka1:9092, group.id: sales-group, auto.offset.reset: largest }).map(lambda x: x[1].decode(utf-8)) def parse_line(line): # 解析一行: timestamp,city,amount try: parts line.split(,) return [(parts[1], float(parts[2]))] # (城市, 金额) except Exception: return [] pairs lines.flatMap(parse_line) # 10分钟窗口2分钟滑动一次 windowed pairs.reduceByKeyAndWindow( lambda a, b: a b, # 窗口内加法 lambda a, b: a - b, # 窗口移出数据时做减法 windowDuration600, slideDuration120 ) windowed.pprint() ssc.start() ssc.awaitTermination()这里提醒一下reduceByKeyAndWindow的第二个参数invReduceFunc是优化用的。如果提供了它Spark 会把新窗口的计算复用旧窗口的部分结果只要增量和减量而不是每次全量计算性能提升明显。但要注意你的聚合函数得满足交换律和结合律保证减出来的结果是正确的。比如算均值就不适合直接这么搞需要额外记录计数和金额总和再相除。3.4 状态管理与 updateStateByKey 的跨批次累积窗口算子解决的是“一段时间内的聚合”但有些业务是“从启动开始至今的累积”典型场景是总访问量、用户总积分。这时你需要跨批次的全局状态。Python 里最常用的状态算子是updateStateByKey。def update_func(new_values, running_state): # new_values: 本批次该 key 的所有新值列表 # running_state: 历史累积的该 key 的状态值 return sum(new_values) (running_state or 0) state_stream pairs.updateStateByKey(update_func) state_stream.pprint()需要注意使用updateStateByKey必须设置 checkpoint否则会直接报错。因为状态更新需要把历史状态持久化只有 checkpoint 才能保存和恢复。checkpoint 目录要放在 HDFS 这类可靠存储上本地目录在分布式环境下会有问题。状态算子用起来顺手但有个坑状态只增不减。像用户离线很久不再活跃他的状态还在内存里占地方运行一两个月后状态数据可能膨胀到拖垮作业。解决办法是结合业务语义做超时清理或者在 update 函数里判断当前时间和最后活跃时间超过阈值就返回None这样对应 key 的状态会被移除。3.5 输出算子和将结果写入外部存储流作业最后必须把结果落出去否则跑完没有任何意义。最常用的输出算子有三个pprint()用于调试控制台foreachRDD()是最灵活的可以拿到每个批次的 RDD 自己写存储逻辑saveAsTextFiles用于快速落盘。foreachRDD是生产代码的主力。我一般把写库逻辑封装在 sink 模块里比如写 MySQL 时控制批量提交条数避免每条数据都建立一次连接。def write_to_mysql(rdd): # 每个批次调用一次 items rdd.collect() if not items: return # 批量 insert注意使用批量参数 # 这里省略连接池和 SQL 细节核心是把 items 一次提交 pass windowed.foreachRDD(write_to_mysql)这里有个细节容易踩坑在foreachRDD里直接创建数据库连接每个 executor 每个批次都会新建连接高并发下连接数会非常吓人。建议使用连接池按 executor 进程维度复用连接而不是每条数据连一次。另外要意识到foreachRDD是在 Driver 端拿 RDD 还是 executor 端拿分区取决于你在哪个层级写代码。如果rdd.collect()到 Driver 再写库数据量大时 Driver 会变成瓶颈如果对每个分区操作就使用rdd.foreachPartition在 executor 端处理性能好很多。4. 生产环境参数调优让作业稳定跑上线4.1 批次间隔、窗口参数如何配合选型批次间隔batchDuration是整个流作业的节拍它同时影响延迟和吞吐。间隔设得太小比如 1 秒处理不过来时任务就会堆积调度器发出延迟警告设得太大比如 5 分钟又满足不了实时性需求。我的选型思路是这样先看业务容忍的最大延迟比如实时大屏要求 30 秒内看到数据那一批最多 30 秒再看数据峰值吞吐量和单批处理耗时如果每小时千万级数据单批 10 秒处理耗时在 5 秒以内那 10 秒就是舒适的节拍。可以通过 Spark UI 观察批处理耗时和调度延迟如果调度延迟持续上升说明批次设置过小或资源不足需要调大间隔或加资源。窗口参数要和批次间隔对齐都取倍数关系。这样窗口边界落在批次边界上状态管理才不会出现错位。我之前接过一个需求批次 5 秒、窗口 60 秒顺手把窗口设成 60结果窗口实际生效和预期差了一个批次的数据最后 debug 了半天才发现窗口必须是批次的整数倍。4.2 背压机制与并行度的关系生产环境最怕的就是数据突发。Kafka topic 平时每秒一万条电商大促时每秒十万条如果 Spark 消费速度跟不上批次处理时间就会越来越长最后整个作业像滚雪球一样积压。这时候背压机制backpressure可以帮我们自动动态调整消费速率。spark.streaming.backpressure.enabledtrue spark.streaming.backpressure.initialRate50000开启后Spark 会根据当前批次的调度延迟和处理耗时动态限制下一次批次的接收速率起到缓冲作用。这里还是要提醒背压只是让系统不崩溃不代表积压的数据会消失。数据还是会在 Kafka 里堆积等峰值过后Spark 之后可以加速消费追补。如果你希望峰值时宁可慢也不能丢这个机制很有效如果你希望保证实时性而可以牺牲一部分吞吐那就别开。并行度方面Direct 方式下每个 Kafka 分区对应一个 RDD 分区所以并行度首先受 Kafka 的分区数限制。如果分区数不够可以先在 Kafka 端扩大分区。其次在 shuffle 类算子执行时要关注spark.sql.shuffle.partitionsSpark SQL 默认值通常较大如果没涉及 SQL则主要看 RDD 默认并行度以及spark.default.parallelism的配置。我踩过的一个坑是Kafka topic 只有 3 个分区但开了 20 个 executor结果单分区数据集中于单个 executor分布式效果大打折扣。4.3 提升吞吐的常用资源配置资源怎么给得结合数据量来判断。一个总的原则是executor 数量乘以单个 executor 可并行 task 数至少要大于 Kafka 总分区数。如果只有一个 partition那开十个 executor 也只有两个 task 在跑取决于核心数。我从一个日均亿级日志的作业里总结了一套还算稳的初始配置具体还是得按你自己的集群规模调配置项设置值说明--num-executors20按数据峰值估算保证并行度足够--executor-cores4单个 executor 并行度对流作业 4 核较稳--executor-memory8g如果频繁 GC可以考虑调大并配合spark.memory.offHeap.enabledspark.streaming.kafka.maxRatePerPartition5000限制单分区消费速率防止峰值冲垮spark.streaming.blockInterval200ms默认 200ms一般无需调整调小可能增加任务数内存方面Python 作业的 executor 内存有一部分要留给 Python worker 进程。PySpark 默认会为每个 executor 启动一个 Python worker实际使用中JVM 侧负责调度和 shufflePython 侧负责 UDF 计算。如果你给 executor 只有 2gPython worker 可能内存不足报出奇怪的 Python OOM。建议在调优时盯着 Spark UI 的 GC 时间和 Python 进程的内存曲线别光看 JVM 那块的指标。4.4 checkpoint 与容错恢复的必备配置生产环境要求作业挂了能自动恢复靠的就是 checkpoint。除了之前说的updateStateByKey必须用 checkpoint所有状态型窗口算子其实都依赖它。checkpoint 会保存两类数据元数据包括 DStream 配置和 RDD 的依赖关系和数据状态快照。它也存在性能消耗如果状态很大checkpoint 的时间会成为瓶颈。对策是适度提高 checkpoint 间隔比如每 5 个批次做一次 checkpoint而不是每个批次。默认行为可以通过参数控制注意不要设置得过长否则状态恢复时回放的数据会很多。ssc.checkpoint(hdfs://namenode:9000/checkpoint/your_app)作业恢复的使用方式spark-submit \ --master yarn \ --deploy-mode client \ your_app.pyDriver 重启后会优先从 checkpoint 目录里恢复元数据和状态而不是重新从 Kafka 读取。这也是为什么 checkpoint 目录必须放在所有 Driver 进程都能访问的存储上。4.5 幂等输出保证不重不丢流处理里有个老大难输出写了两遍怎么办因为流作业可能因为故障重启重启后某些批次可能重复处理如果你写的是累加型的 SQL比如INSERT INTO stats SELECT count(*) FROM ...重复执行一次数值就翻倍了。解决方法说白了就是让输出操作具有幂等性。写 MySQL 时用唯一键冲突时更新写 Hive 时用分区覆盖每次基于批处理时间写同一个分区目录写 Redis 时用 SETNX 或固定 key 覆盖。关键是所有输出操作要能承受重复执行而不产生脏数据。如果依赖外部事务要准确把握事务的提交时机我见过在检查点恢复后重复调用外部事务导致数据重复入库的情况排查起来非常痛。5. 实战踩坑与问题排查实录5.1 坑一批次调度延迟越来越大最终积压到无法恢复现象Spark UI 里 Scheduled Delay 从十几秒一路涨到几分钟程序还在跑但输出的数据一直落后于当前时间。我当时排查的思路先看是不是批次处理时间超过了批次间隔。如果单批耗时长优先看资源 CPU 和内存是否有瓶颈再看 GC 时间是不是太长如果都没有问题再检查是不是有数据倾斜某个分区数据量特别大拖慢了整体。那次最终定位到问题出在一张维表 join 上。我用 broadcast join 广播了大维表但表每天更新我没刷新导致 Shuffle 数据量巨大。换用动态广播或者改成编码后本地维度匹配之后调度延迟就降下来了。这类问题最怕的是没头绪我的排查顺序一般是从 Spark UI 的 Batch Processing Time 曲线看起再逐项排除资源、GC、倾斜、算子问题。5.2 坑二累积状态数据无限膨胀内存爆了用了updateStateByKey累积用户积分跑了三个月之后executor 不断出现内存溢出重启也没用过几天又爆。原因很简单旧的 key 永远不会过期每个 key 的状态一直躺在内存里。用户可能是几百万级别每个 key 状态字段不少内存就被耗尽了。解决办法是在 update 函数里引入状态清理逻辑基于最后活跃时间判定是否过期。具体做法是状态里保存一个 lastTimestamp每次更新逻辑里判断当前时间 - lastTimestamp是否超过阈值超过则返回None让 Spark 把该 key 从状态表里移除。在这里补充一个调优经验也可能用mapWithState替代updateStateByKey它在性能上强于后者但 Python API 里支持的便捷程度不如 Scala我并没有在线上真正使用过。所以如果你的状态规模大最重要的不是换 API而是合理清理。5.3 坑三Kafka offset 处理不当导致重复消费或丢失有一次作业重启后我对比了两个数据源的记录数发现消费端的记录比生产端多出来不少明显是重复消费。Direct 方式下Spark Streaming 会周期性地把当前批次的消费 offset 保存到 checkpoint 里。但 checkpoint 和实际输出不是原子的。如果数据已经写入 MySQL但 checkpoint 还没保存作业崩溃重启后就会从旧 offset 重新消费重复写 MySQL。如果相反checkpoint 保存了但 MySQL 写入在事务回滚里没完成又会丢数据。最稳妥的方案是把 offset 管理彻底外部化。每次批次处理完成后把当前批次的 offset 提交到 Kafka 自身通过 commitSync、或者保存到 Zookeeper/数据库。这样恢复时从外部记录的位置消费不依赖 checkpoint。配合上一节说的幂等输出可以做到端到端不重不丢。这个过程实现起来有点繁琐但线上可靠性的价值远超那点开发成本。5.4 坑四Python UDF 性能瓶颈作业吞吐上不去场景是这样的每条 Kafka 消息需要调用一个外部接口做内容识别再把结果打标输出。我在 map 里直接调用了外部 HTTP 接口结果吞吐惨不忍睹单批次处理时间从几十秒暴涨到几十分钟。原因很直接每条消息同步远程调用网络等待时间远大于计算时间而 Spark 的 map 算子是并行执行的但外部接口的响应时间成了最大瓶颈。解决方案有两个方向第一个方向是减少外部调用频次把单条调用改成批量调用。比如用mapPartitions在每个分区内把一批消息攒起来一次性发到外部接口做批量识别接口支持批处理的话效果立竿见影。第二个方向是增加并行度把批次的数据重新分区虽然自定义分区逻辑要小心但可以让更多 executor 同时发起请求。这里还有个更隐蔽的点Python 的普通函数在 PySpark 里执行每条数据都要经过 JVM 和 Python 进程之间的序列化和网络传输本身开销就不小。如果处理函数非常简单比如只是简单的字段拼接这种开销会占比很高。此时可以考虑把纯改动逻辑用 Spark SQL 的内置函数实现尽可能用 JVM 原生的算子处理减少 Python 参与的数据量。5.5 常见问题速查表现象可能原因推荐排查路径作业启动后长时间不输出本地模式只给了 1 个线程没有线程处理数据用local[2]或以上启动调度延迟持续上升批次处理耗时超过批次间隔资源不足或计算过重看 Batch Processing Time调整资源或间隔状态算子报错 checkpoint directory has not been set使用了状态算子却没设置 checkpoint调用ssc.checkpoint(...)重复消费数据offset 保存与输出不一致依赖了 checkpoint 恢复外部化管理 offset输出改成幂等写频繁 OOM并行度过高、状态膨胀、或 Python worker 内存不足调并行度、加状态清理、调大 Python worker 内存数据倾斜导致单批很慢key 分布不均单个分区数据量巨大加盐重分区或先对热点 key 单独处理作业因 driver 网络闪断挂掉没有您想象的那么罕见需要自动重启机制部署 supervisor 或 YARN 自动重启配置 checkpoint我在多次排查中养成了一个习惯所有流作业上线前必须压测验证的不是数字好不好的问题而是弄清楚“当数据量以 10 倍峰值打进来时系统的行为是什么”。背压会不会触发、批次会不会积压、外部存储扛不扛得住这些问题在压测阶段暴露出来远比线上故障时排查要快得多。6. 写到最后聊点实际的我个人在这套技术栈上摸爬滚打下来最大的体会是Spark Streaming 绝对不是最潮的流计算框架但它在数据量大、业务链路复杂、依赖 Spark 生态的场景下依然是一个非常务实的选择。Python 版虽然有一些性能上的坑但只要选对算子、做好参数调优完全能支撑大规模生产流量。如果你现在维护的老作业还是 DStream API也不用急着推翻重写先把状态清理、offset 外部化、输出幂等这三件事做好系统的稳定性会立刻上一个台阶。最后分享一个小技巧本地调试时数据源不好模拟的话直接用 Kafka 的 producer 脚本往 topic 里灌测试数据再用一个小脚本监听程序输出。环节越接近生产你发现环境问题的概率就越大。如果你正打算把一个批处理作业改造成实时作业或者被线上流作业的延迟问题困扰可以按本文的顺序做一遍排查和调参大部分问题都会浮出水面。祝你的数据流跑得又稳又快。