1. 项目背景与技术选型逻辑
1.1 这个项目解决的是什么问题
先说清楚这套技术栈到底在解决什么。传统的大数据处理链路，十有八九是离线批处理这套逻辑：业务数据白天产生，晚上用调度系统把当天的数据捞出来，丢给Hive或者Spark跑一批SQL，第二天早上看报表。这套流程在数据量没上来、业务对时效性不敏感的时候，完全够用。但一旦业务方开始提"我要看实时的转化漏斗""用户点击后几秒内就要触发推荐更新"，离线那一套就彻底顶不住了，因为它的响应单位是"天"而不是"秒"。
Kafka + Spark Streaming这套组合，要解决的就是这个"秒级到分钟级延迟"的中间地带。它不需要像真正的毫秒级流处理那样去裸写网络层，也不像离线批处理那样要等一整个调度周期。Kafka负责把源源不断产生的数据稳定地收进来、存起来、按序喂给下游，Spark Streaming则用微批（micro-batch）的思路，把连续不断的流切成一小块一小块的RDD，然后用Spark生态里成熟的计算引擎去处理。简单说，这套组合的核心价值就是：用一个团队已经熟悉的批处理思维，做出接近实时的处理效果。
我最早接触这套东西，是在一个用户行为分析的项目里。当时的需求是实时统计不同渠道过来的注册转化率，渠道投放那边催得特别急，说晚一个小时看到数据，预算就多烧一个小时。但团队里没人写过真正的流处理代码，Storm那时候已经快过气了，Flink又没人敢碰，最后选来选去，落在了Spark Streaming上。原因很直接：我们已有的离线ETL全是Spark SQL，团队对Spark的算子、调优手段已经滚瓜烂熟，把一套批处理的经验平移过来，学习成本是最低的。这也是一直到现在，我依然认为Spark Streaming是个"性价比极高"的入门实时计算框架的原因。
1.2 为什么是Kafka + Spark Streaming这对组合
很多人会纠结一个问题：既然要做实时，为什么不用Flink？这个问题换个角度回答更实际：不是每个项目都需要毫秒级的精确一次性语义。如果你的场景是"秒级延迟能接受、数据量在每秒几千到几万条、团队已经有Spark经验"，那Spark Streaming就是更务实的选择。Kafka + Spark Streaming这对组合的江湖地位，本质上是由Kafka的"缓冲区"属性和Spark Streaming的"微批"属性天然互补决定的。
Kafka在这条链路里的角色，不只是消息队列那么简单。它是一个分布式的、带持久化的提交日志。这意味着数据一旦写入Kafka，就算下游计算程序挂掉或者重启，消息也不会丢，只要offset没丢，重启后可以接着上次的位置继续消费。这个特性对流式计算来说太关键了，它让整个链路有了一个可以回溯、可以重放的"数据保险箱"。
而Spark Streaming做的事情，是把Kafka里的数据按时间窗口切成一个个小批次，每个批次就是一个RDD，接下来你就可以用map、flatMap、reduceByKey这些熟悉的操作去处理它了。从我实际使用的感受来说，这套组合最大的好处是"每一环都有兜底"：Kafka兜住了数据不丢，Spark Streaming的checkpoint兜住了计算状态不丢。对于没有太多流计算经验的中小团队，这种"处处有保底"的设计，比Flink那种精密的精确一次语义更容易驾驭。
2. 环境准备与版本兼容避坑指南
2.1 组件版本选型：这一步定生死
我必须把版本兼容性放在实操第一步来讲，因为这一块踩坑的代价实在太大。Spark Streaming和Kafka的集成，不是简单地把两个最新版本拼在一起就能用的。常见的坑包括：Kafka的broker端协议版本和客户端版本不匹配、Spark Streaming的Kafka依赖包把自己锁在某个Scala版本上、以及Kafka 0.10之后consumer重balance机制变化导致的代码写法完全不一样。
我个人的经验是，先定Spark版本，再倒推Kafka版本。为什么这样定？因为Spark Streaming的Kafka集成依赖（spark-streaming-kafka-0-10）是和Spark本身一起发布的，它的Scala版本、编译方式和Spark核心绑定很紧。你选了一个很新的Kafka版本，但对应的Spark集成库还没有跟进，就只能等社区更新或者自己踩兼容性的坑。反之，先选一个社区验证充分的Spark版本，然后用这个版本适配的Kafka版本，通常能避开绝大多数问题。
这里直接分享一套我实测过比较稳的搭配：Spark 2.4.x配Kafka 2.x，Scala用2.11或者2.12都行，但要注意Kafka的客户端jar包版本要和broker端保持一致。另外，Java版本必须用Java 8，如果你用了Java 11，Spark 2.x的某些反射调用会报模块访问错误。
从搭建层面讲，建议先在本地起一套测试环境。Kafka不用装集群，单机模式就够了，重点是把broker、zookeeper（或者KRaft模式下的metadata）和topic创建这套流程跑通。Spark也不用上集群，本地模式（local[*]）足够用来调通整个消费链路。
2.2 任务目标：先定义清楚"实时"的含义
在动手写代码之前，我强烈建议你先跟业务方确认一个事：你说的"实时"，到底是多久？这个问题听起来很简单，但在实际项目里，业务方的"实时"可能是秒级，也可能是"几分钟内能看到就行"。搞清楚这个边界，直接决定了你的批处理窗口设多大、用不用状态算子、要不要上窗口计算。
举个例子。我之前做的那个转化率统计项目，业务方最开始说"实时统计"，我们按5秒的批次间隔把架构搭起来了，结果后面发现业务方其实是希望"投放人员刷新页面时，看到的数据延迟不超过1分钟"。这种情况下，把批次间隔从5秒放宽到30秒，性能压力能小好几倍，代码完全不用动。所以，先定义清楚SLA（服务等级协议），再去做技术设计，这永远是第一步。
这个SLA还会影响你对其他组件的选型。如果你的容错目标是"极端情况下最多丢几秒的数据"，那生产端可以不开acks=all，性能能高不少；如果要求一条都不能丢，那acks=all加上重试机制就是必须的。这些都应该是设计阶段定下来的，而不是上线之后被业务方追问才去补。
具体到本项目的目标，可以定义成一套比较标准的实时数仓雏形：数据从业务系统产生，由生产端写入Kafka，Spark Streaming实时消费，做清洗、转换、聚合，把结果写入下游的存储（比如MySQL、Redis、ClickHouse），最终支撑一个实时看板。这是一个哪怕没有任何流计算经验的人，也能通过这套链路完整理解实时计算全流程的经典架构。
3. Kafka端的关键设计与生产端实践
3.1 Topic设计：从源头控制好数据质量
Kafka的Topic设计，是整条链路的源头。这里的核心决策包括：topic数量怎么规划、分区数设多少、副本数怎么配、key怎么设计。这些决策一旦做下去，后面改动成本极高，所以必须在一开始就规划好。
先聊分区数。Kafka的分区数是并行度的上限，Spark Streaming消费某个topic时，能启用的最大并发consumer数就等于分区数。你设了6个分区，那最多就只能有6个consumer线程同时拉数据。反过来，分区数也不是越多越好。分区越多，broker端的文件句柄和内存开销越大，consumer端的rebalance时间也越长。我的经验是，结合下游处理能力和数据量来定，一般业务场景下，单topic分区数在6到12之间是个不容易出错的区间。
然后是key的设计。Kafka的key有两个用途：一个是决定消息进哪个分区（相同key的消息会进同一个分区，进而被同一个consumer处理），另一个是作为消息内容的业务主键，用于去重、排序。在实时计算场景里，key选得好，很多问题能自动消失。比如你要按用户维度做实时统计，那key就选userId；要按订单维度处理，key就选orderId。如果你用随机key，数据会均匀分布到所有分区，但相同用户的消息会被打散到不同分区，你要做用户维度的聚合就麻烦了。
最后是副本数和数据保留策略。副本数设2到3就够了，太多会拖慢写入性能；保留策略看你的下游有没有重放需求，如果Kafka只是临时缓冲，保留1天甚至几个小时都够，如果还要承担离线数仓的数据源角色，那保留时间就得按天算。
3.2 生产端的可靠性与吞吐权衡
Kafka生产者的参数配置，是Kafka + Spark Streaming链路里最容易被忽视但又影响巨大的环节。核心就一句话：要在数据可靠性和写吞吐量之间找一个平衡点。
生产端的三个关键参数是acks、retries和linger.ms。先说acks，它有3个取值：acks=0表示发送出去就算成功，不等待任何确认，吞吐最高，但broker挂了就丢数据；acks=1表示等待leader副本写入成功就算成功，这是吞吐和可靠性的中庸选择；acks=all表示等待所有ISR副本都写入成功才算成功，最安全，但延迟明显上升。如果你做的是用户行为采集这类"丢了也能忍"的数据，acks=1足够；如果是订单、支付这类核心业务数据，我建议直接acks=all，别在这上面省。
linger.ms这个参数，作用是让生产者攒一批消息再发出去，默认值是0，也就是来一条发一条。很多人为了降低延迟，把linger.ms保持默认，但实际上在每秒几千条的写入场景下，稍微把linger.ms调到5到20毫秒，批量发送能显著提升吞吐，延迟增加的那几毫秒对于实时计算链路完全无感。还有一个batch.size参数，默认16KB，如果你单条消息比较大，建议调大到64KB甚至更高，否则批次经常装不满就发出去，linger.ms的效果就被削弱了。
我个人的建议配置是：acks=all（核心数据）或acks=1（非核心数据），retries设成3到5，linger.ms设成10，buffer.memory调大到64MB以上。这套配置我实测过，在每秒1万条左右的消息量下，写入稳定且磁盘占用可控。注意retries设大了之后，有可能因为重试导致消息重复写入，这时需要让消费者端具备幂等处理能力，或者开启Kafka的幂等生产者（enable.idempotence=true），后者能保证生产者侧不产生重复消息。
4. Spark Streaming消费端核心实现
4.1 两种接入方式的选型：Receiver还是Direct
Spark Streaming消费Kafka，有两种历史遗留的接入方式，如果你在网上搜教程，会发现很多老教程还在讲Receiver方式，这里我要明确说：直接用Direct方式，不要用Receiver方式。这不是口味问题，而是架构层面的重大差异。
Receiver方式是把Kafka的数据先拉到Spark的executor里，存进WAL（预写日志），再被Spark Streaming消费，这中间多了一层缓冲，带来两个问题：一是消息可能重复，因为WAL重放会导致数据被处理两遍；二是executor挂了之后，这块缓冲区的数据会丢失。最要命的是，Receiver方式的offset是存在ZooKeeper里的，Spark Streaming实际上以ZooKeeper里的offset为准，这就可能出现Spark处理和ZK记录不一致的情况。
Direct方式（自Spark 1.3引入）直接把Kafka的partition和Spark Streaming的partition一一对应起来，不再有Receiver那层缓冲。Spark的executor直接从Kafka的partition拉数据，数据不落中间层，offset由Spark自己维护。这样带来的好处很明显：精确一次语义在配合手动管理offset时可以做到，不会无故丢数据，而且因为去掉了Receiver层，执行效率更高、资源占用更少。
唯一的注意点：Direct方式依赖Kafka的API版本，必须用spark-streaming-kafka-0-10这个依赖，而它只支持Kafka 0.10及以上版本。如果你还在用老掉牙的Kafka 0.8、0.9，那就只能被迫用Receiver方式，但这种古董环境本身就应该尽早升级了。
4.2 Direct模式的核心代码与关键参数
讲完选型，直接给一套能跑的Scala代码。这套代码是Direct模式的标准写法，目标是从Kafka拉取用户行为日志，解析成结构化字段，按用户维度和事件类型做实时聚合。
import org.apache.kafka.common.serialization.StringDeserializer import org.apache.spark.SparkConf import org.apache.spark.streaming.kafka010._ import org.apache.spark.streaming.{Seconds, StreamingContext} object KafkaStreamingApp { def main(args: Array[String]): Unit = { val sparkConf = new SparkConf() .setAppName("KafkaStreamingApp") .setMaster("local[*]") .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") val ssc = new StreamingContext(sparkConf, Seconds(5)) // 开启checkpoint，用于故障恢复和状态计算 ssc.checkpoint("hdfs:///tmp/streaming-checkpoint") val kafkaParams = Map[String, Object]( "bootstrap.servers" -> "node01:9092,node02:9092", "key.deserializer" -> classOf[StringDeserializer], "value.deserializer" -> classOf[StringDeserializer], "group.id" -> "user-behavior-group", "auto.offset.reset" -> "latest", "enable.auto.commit" -> (false: java.lang.Boolean) ) val topics = Array("user-behavior-log") val stream = KafkaUtils.createDirectStream[String, String]( ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) ) // 解析消息体（假设JSON格式） val parsed = stream.map(record => { val json = record.value() // 这里用你最顺手的JSON解析库，fastjson、jackson都可以 // 示例：从JSON中提取 userId, eventType, ts (userId, eventType, ts) }) // 按 (userId, eventType) 做实时计数 val counts = parsed .map { case (userId, eventType, ts) => ((userId, eventType), 1L) } .reduceByKeyAndWindow(_ + _, Seconds(60), Seconds(10)) counts.print() // 启动流处理 ssc.start() ssc.awaitTermination() } }代码本身不复杂，但有几个参数值得展开讲讲。
Seconds(5)是批次间隔，这也是Spark Streaming里最核心的一个参数。设得太短，比如1秒，会频繁调度任务，集群的调度开销会被放大，而且每个批次的处理都要在下一个批次到来前完成，否则会堆积；设得太长，实时性就差。我踩坑后的建议是：先按业务能接受的最低实时性要求（比如10秒）起步，观察处理后端压力再逐步调短。
enable.auto.commit一定要设为false，然后手动管理offset。这是实现精确一次语义的前提。如果你开着自动提交，Spark会在拉取消息后自动提交offset，但此时这批数据可能还没处理完，甚至处理失败了，消息就丢了。手动提交的时机应该放在——这批数据真正处理完成、结果写入了下游之后。
这里要插入关于手动提交offset的最佳实践：不建议在每批次处理完就立即提交，因为如果下游写入失败，offset已经提交了，重启后这批数据就被跳过了。更好的做法是把offset的提交和下游写入做成"同事务"或"至少一次"的补偿关系——比如写下游之前先记录这批数据的offset，下游写入成功后再提交offset。我用Redis做过这个逻辑：批量处理完后，把(partition, offset)写入Redis，然后在Kafka consumer里初始化offset时从Redis读取。这样就算程序意外崩溃，重启后也能从上一次成功处理的位点继续。
4.3 窗口计算与状态管理的组合用法
单个批次内的聚合（reduceByKey）只能统计当前这个微批里的数据。但实时计算里大量需求是"最近N分钟的用户活跃数""最近1小时每个页面的PV"，这就必须用窗口计算。（reduceByKeyAndWindow 或 window 方法）
窗口计算有两个参数，一个是窗口长度（window length），表示你要统计多长一段时间的数据；另一个是滑动间隔（slide interval），表示每隔多久输出一次结果。比如窗口长度60秒、滑动间隔10秒，意思就是每10秒输出一次最近60秒的汇总数据。这两个值都必须是批次间隔的整数倍，不然报错。
用窗口计算要特别注意性能：窗口越长，需要维护的中间状态越多。我之前遇到过一个问题，窗口设成2小时，滑动间隔10秒，结果每个批次都要重新计算2小时内的全部数据，集群负载直线上升。后来把批处理逻辑改成增量式的，用updateStateByKey维护历史状态，每次只计算新进来的数据，状态更新到旧的累计值上，负载才降下来。如果你用的是Spark 2.4以上的版本，更推荐用mapWithState，它的状态管理效率比updateStateByKey高很多，但写法上要兼容好Scala的版本。
5. 性能调优的实操参数：让集群跑得更稳
5.1 批次任务内存与GC调优
Spark Streaming的每个微批，本质上是一次小的Spark作业，它会启动一批Executor来执行任务。在长时间运行的流任务上，最容易出问题的就是内存和GC。具体表现是：任务跑几天之后，GC时间越来越长，批次处理时间逐渐拉长，甚至出现OOM。
这个问题的根源，在于Spark Streaming的批次数据在内存中长期驻留，尤其是开启了窗口计算和状态管理之后，老年代里堆着大量历史状态对象，Full GC频繁触发，STW（stop-the-world）时间拖垮整个链路。
我的经验是这么几个手段。首先，JVM堆内存不要贪大。很多人觉得内存越大越好，堆到32GB，结果GC一次要停顿好几秒，反而比8GB堆更卡。在流计算场景下，我建议单Executor的堆内存控制在8GB到16GB之间，配合G1垃圾回收器，把-XX:MaxGCPauseMillis设到200ms左右，让GC停顿时间变得可接受。其次，开启Kryo序列化，这能显著降低对象大小，减少GC压力。spark.serializer设置为KryoSerializer，同时注册需要用到的类，能有效减轻GC负担。第三，合理设置spark.streaming.kafka.maxRatePerPartition，限制每个分区每秒拉取的最大消息数，避免某个瞬间Kafka堆积了大量数据导致批次数据量暴增，冲垮内存。
5.2 背压机制：让系统自己刹车
背压（backpressure）这个概念，简单说就是当数据涌入速度超过处理速度时，系统自动减速，避免自己被冲垮。没有背压机制的情况下，如果某个时段数据流量突然暴增3倍，Spark Streaming会硬着头皮把积压在Kafka里的数据全部拉进来处理，结果就是这批任务的执行时间远超批次间隔，造成严重堆积，延迟从秒级恶化到分钟级甚至小时级。
Spark Streaming从1.5版本开始支持背压，是通过一个叫RateController的组件实现的，它根据最近几个批次的调度延迟和处理时间，动态调整每个分区每秒拉取的最大消息数。你要做的就两件事：把spark.streaming.backpressure.enabled设置为true，同时把spark.streaming.kafka.maxRatePerPartition设置成一个合理的上限，这个上限作为背压控制的初始值。
我实际项目里的调法是：先按正常流量的1.5倍估算一个maxRatePerPartition，开着背压跑几天，观察平均批次处理时间和调度延迟，如果调度延迟长期存在，说明处理不过来，就再往下调maxRate；如果批次处理时间稳定小于批次间隔，可以考虑往上调。这里的经验是一个动态调整的过程，不要指望第一次设好就一劳永逸。不需要关闭背压，除非你有极强的数据突发处理需求并且准备了充分的资源冗余。
5.3 并行度与资源分配的权衡
并行度是另一个很容易拍脑袋设置的参数。Spark Streaming里的并行度，包含几个不同层级：一个批次数据被分成多少个partition、每个partition的数据量多大、用多少个Executor核来跑。如果每个分区数据量太大，任务执行时间过长；如果分区太多，任务调度开销过大。
我的经验是，按每个任务处理5MB到10MB数据来倒推并行度。假如一个批次有100MB数据，那就设10到20个分区。在Spark Streaming里调整分区数的一个常用手段是repartition：如果需要增大并行度，直接对DStream调用repartition(更大的数值)就能让后续的计算分布到更多任务上。如果Kafka这边分区数不够，那就要在topic设计阶段预判好流量增长，把topic分区数设够。
还有一个容易忽视的点：Executor的资源与Kafka分区数的对应关系。如果你有6个Kafka分区，但只给了2个Executor核，那最多同时2个分区在消费，其他4个分区在排队，实时性会打折扣。理想情况下，消费端的Executor核数应该不小于Kafka分区数，让每个分区都有一个独占的任务在拉取数据。
6. 踩坑记录与常见问题排查实战
6.1 重复消费与offset管理的实战处理
使用Kafka + Spark Streaming，你一定会遇到的一个问题是重复消费。先说结论：在分布式系统里，做到严格精确一次语义，需要多个组件配合，代价很高；对于大多数实时计算场景，"至少一次"语义加上业务侧幂等，是性价比最高的方案。
重复消费的典型场景是：某个批次的数据已经处理完了，下游结果也已经写入了，但在offset提交之前，程序突然崩溃（比如OOM被杀、网络断开、手动kill）。重启后，Spark从上次提交的offset继续消费，刚才那批数据就会被再处理一遍。如果你下游的写入操作不是幂等的（比如直接insert数据库记录），就会出现重复数据。
我踩过几次坑之后，总结出的处理套路是：下游写入做成幂等。比如写MySQL时，用唯一索引（比如业务主键+事件时间）做去重，重复插入会被数据库挡住；写Redis时，用SET覆盖写天然幂等；写ClickHouse时，可以在表上建ReplacingMergeTree，利用版本号去重。如果业务上无法容忍任何重复，那就需要手动管理offset并在事务内提交，比如将处理结果和offset写入同一个数据库事务，但这块复杂度会明显提升，需要在项目早期就规划好。
6.2 数据倾斜：一个热门key引发的雪崩
数据倾斜这个问题，离线批处理会遇到，流处理同样会遇到。而且流处理的实时性要求让倾斜问题的放大效应更明显，因为某个key的数据一旦堆积，这个批次处理不完，下一个批次还在往里灌，瓶颈会持续恶化。
一个典型场景：做用户行为统计时，"热门用户"产生的消息量可能是普通用户的几千倍。用reduceByKey聚合时，这个热门user的key所在的partition，处理时间可能是其他partition的几十倍，拖慢整个批次的完成时间。反映到监控上，就是某个executor的GC特别频繁、某几个任务反复失败，而其他任务早就跑完了。
对于数据倾斜，我处理过的方案里，最有效的是"两阶段聚合"。以WordCount式的计数为例，第一阶段给key加随机前缀，比如key+random(1-10)，先做一次局部聚合，把数据分散到更多分区上；第二阶段去掉前缀，再对局部聚合结果做一次全局聚合。这样原本集中在一两个task上的计算压力，被均匀分摊到了10个task上。代价是多了几秒钟的延迟和一轮shuffle，但相比被一个task卡死，这点代价完全值得。
还有一种情况是倾斜发生在源端。比如Kafka的某个分区堆积了大量消息，其他分区都消费完了，只有这一个分区还在慢慢拉。如果你的key设计得不均匀（比如按用户ID哈希），就会出现某个分区永远比其他分区数据量大。这时候要么把这个topic的分区数加大并重新分区，要么把数据打散后在下游做二次聚合。
6.3 checkpoint的陷阱：不是开了就万事大吉
Spark Streaming的checkpoint，是很多教程里一笔带过但实际暗坑很多的地方。它的设计目的，是让流计算程序在宕机后能从最近的检查点恢复，包括配置、DStream操作图和RDD数据三部分。看起来完美，实际用起来有几个坑。
第一个坑，checkpoint的数据结构跟代码强绑定。如果你改了代码逻辑（比如改了处理函数的逻辑、改了算子链），重启后从checkpoint恢复时会反序列化旧的DStream图，这跟你新代码的结构经常对不上，直接报序列化错误或者行为不符合预期。很多初学者遇到这个问题，把checkpoint目录删了重启才"修好"，但代价是状态也一起丢了。我的经验是：代码结构变了，就换一个新的checkpoint目录，或者干脆手动清除旧的checkpoint，然后从Kafka的offset重新消费，这样虽然会丢掉窗口历史状态，但至少是干净确定的。
第二个坑，把checkpoint目录放在本地磁盘上，对于分布式集群环境，这等于没设。因为每个executor落在不同机器上，本地磁盘的checkpoint互相不共享，driver挂了根本无法恢复。建议统一放到HDFS或者其他分布式文件系统上，保证跨节点可达。
第三个坑，checkpoint的写入频率。默认是在每个批次处理完成后写入，也就是说，状态越大、批次间隔越短，checkpoint写入越频繁，对HDFS的压力越大。我试过在窗口很长、状态很大的场景下，checkpoint写入成了瓶颈，每批次的执行时间从几秒变成几十秒。解决办法是把基于RDD的checkpoint换成基于状态的checkpoint（mapWithState里有个checkpointDuration参数可以调），或者把HDFS的副本系数降为2，减轻写入压力。
6.4 任务堆积导致延迟飙升的排查思路
最后整理一个排查"批次堆积延迟飙升"的实战思路，这类问题的排查路径相对固定，先看容易忽略的环节，其次重点检查状态和GC情况。
首先去Kafka侧看消费组的Lag（消费积压）。用kafka-consumer-groups.sh --describe --group 消费组名，能看到每个分区的当前offset和log-end-offset。如果Lag在持续增长，说明消费速度跟不上生产速度，问题在下游处理端；如果Lag没有增长，说明消息是稳定消费的，延迟可能是消费端调度或下游写入的问题。
下一步看Spark UI的Streaming页面，重点关注两个指标：Scheduling Delay（调度延迟，批次进入调度到实际执行的时间差）和Processing Time（处理时间）。如果Processing Time持续大于批次间隔，说明这批数据没处理完下一批就来了，任务开始堆积。这时候就要看具体是哪个stage慢，点进去看每个task的执行时间分布，确认是不是有倾斜的task。
如果看到某个task的GC时间特别长，优先检查状态数据是不是太大了，或者是不是堆内存不够；如果task长时间卡在某个Kafka partition的拉取上，检查那个partition在Kafka侧的文件段是不是特别大，影响了读取效率。我之前遇到过一次，Kafka的log.segment.bytes设得太小，生成了海量小文件，Spark拉取时大量磁盘随机IO，速度快不起来，后来把segment调大之后明显改善。
整个排查的思路就是：从Lag判断问题在前在后，从Processing Time判断计算是否健康，从task级别找具体瓶颈，从GC和IO状态定位最终病灶。这样一条线走下来，大多数堆积问题都能在半小时内定位到根因。
最后补充一点，这个项目后面如果要继续扩展，比较自然的方向是：把Spark Streaming的DStream API迁移到Structured Streaming，或者把窗口统计结果下沉到ClickHouse做自助分析，又或者引入Kafka的KSQL做纯实时ETL。但不管往哪个方向走，Kafka + Spark Streaming这套组合打下的"数据链路认知"——数据怎么进、怎么攒、怎么算、怎么落地——都是通用的，这才是这套架构给一个工程师留下的最值钱的东西。
