026、流式计算：Kafka与Spark Streaming实时处理

一、深夜告警数据延迟了15分钟上周四凌晨两点手机突然震个不停。监控系统告警实时推荐引擎的数据流水线延迟超过15分钟。打开监控面板Kafka消费者组的lag曲线像爬山一样往上窜Spark Streaming的批处理时间已经超过窗口间隔。这不是第一次了但这次特别棘手——业务方要求99.9%的消息必须在5秒内处理完毕。先查了Spark UI发现几个executor的GC时间占到了处理时间的40%。再翻Kafka consumer日志看到频繁的rebalance记录。问题逐渐清晰数据倾斜遇上配置不当再加上序列化开销整个管道就像早高峰的十字路口堵死了。二、Kafka不只是个消息队列很多人把Kafka当成加强版RabbitMQ用这是第一个坑。Kafka的核心是分布式提交日志它的设计目标是大规模、高吞吐的流数据持久化。// 错误示范每次消费都提交offset性能杀手consumer.poll(100).forEach{recordprocess(record)consumer.commitSync()// 别这样写同步提交每个消息}// 建议写法批量处理异步提交valrecordsconsumer.poll(Duration.ofMillis(1000))batchProcess(records)consumer.commitAsync()// 这里可以加回调处理提交失败分区策略是关键。曾经有个项目所有数据都发到同一个分区因为key设成了常量。下游Spark任务只有一个task在干活其他都在围观。// 分区数估算公式经验值// 目标吞吐 单分区吞吐 * 分区数// 单分区吞吐经验值机械盘~10MB/sSSD~30MB/s// 建议分区数 峰值吞吐 / 单分区吞吐 * 1.2预留buffer三、Spark Streaming微批处理的陷阱Spark Streaming不是真正的流处理而是微批处理。这个本质决定了它的行为模式。// 窗口操作容易踩的坑valstreamssc.socketTextStream(...).map(...).window(Minutes(10),Minutes(5))// 窗口长度10分钟滑动间隔5分钟// 注意这里每个窗口包含的数据量是变化的// 如果上游数据突发OOM就来了checkpoint目录配置不当会导致任务重启失败。曾经有同事把checkpoint放在HDFS默认路径没设清理策略最后磁盘写满整个集群瘫痪。ssc.checkpoint(hdfs://path/to/checkpoint)// 一定要配置自动清理比如sparkConf.set(spark.cleaner.ttl,3600)背压机制backpressure要打开不然数据洪峰时直接打垮executorsparkConf.set(spark.streaming.backpressure.enabled,true)sparkConf.set(spark.streaming.kafka.maxRatePerPartition,1000)// 每分区最大速率四、Kafka Spark Streaming集成实战两种集成方式Receiver-based和Direct模式。现在基本都用Direct模式createDirectStream它更简单也更容易保证Exactly-Once语义。valkafkaParamsMap(bootstrap.servers-broker1:9092,broker2:9092,key.deserializer-classOf[StringDeserializer],value.deserializer-classOf[StringDeserializer],group.id-spark-streaming-group,auto.offset.reset-latest,// 生产环境建议earliest避免丢数据enable.auto.commit-(false:java.lang.Boolean)// 必须关掉让Spark管理offset)valtopicsArray(user_behavior)valstreamKafkaUtils.createDirectStream[String,String](ssc,PreferConsistent,Subscribe[String,String](topics,kafkaParams))// 手动维护offset到外部存储如Redis、MySQLstream.foreachRDD{rddvaloffsetRangesrdd.asInstanceOf[HasOffsetRanges].offsetRanges// 处理业务逻辑processRDD(rdd)// 提交offsetoffsetRanges.foreach{rangesaveOffsetToStorage(range.topic,range.partition,range.untilOffset)}}这里有个细节offset提交要在业务逻辑成功之后但又要保证原子性。我们曾经遇到过处理成功但offset提交失败导致数据重复消费。后来引入了事务表才解决。五、调优笔记从血泪教训中总结序列化用Kryo别用Java原生序列化。配置时记得注册自定义类sparkConf.registerKryoClasses(Array(classOf[UserEvent],classOf[Order]))并行度Kafka分区数和Spark分区数最好保持1:1或整数倍关系。曾经设了60个Kafka分区Spark却只有10个core资源浪费严重。内存管理Streaming应用对内存敏感建议单独设置executor内存模型--executor-memory 4G\--confspark.executor.memoryOverhead1024\--confspark.streaming.unpersisttrue# 自动清理已用过的RDD监控指标除了系统监控一定要埋业务层的延迟统计。我们曾经系统指标一切正常但业务延迟很高最后发现是外部API调用超时。失败恢复测试各种失败场景——Kafka broker重启、Spark executor挂掉、网络分区。有个经验先停掉Spark任务往Kafka灌一批数据再重启任务检查offset是否从正确位置开始。六、经验之谈流式系统的哲学流处理系统不是批处理的加速版而是另一种编程范式。最大的思维转变是从关注数据全集到关注数据变化。调试流式作业不要只盯着代码。去看Kafka的监控指标ISR数量、网络出入流量、Spark的GC日志、操作系统的IOwait。很多时候问题不在应用层。生产环境一定要有熔断机制。我们现在的方案是当延迟超过阈值自动切换到一个降级处理路径比如跳过复杂特征计算保证数据至少能流动起来。最后流处理系统的复杂度是阶跃式上升的。从demo到POC再到生产每跨一个阶段要考虑的问题多一个数量级。建议循序渐进先保证数据能流起来再优化延迟最后追求Exactly-Once语义。别想一步到位那会掉进无数个坑里爬不出来。七、写在最后实时流处理就像维护一条高速运转的流水线任何一个环节的微小卡顿都会累积成整个系统的延迟。最好的学习方式不是读文档而是亲手搭一套环境然后模拟各种故障场景。保持对数据的敬畏——它不会按你设想的方式到来总是在你最困的时候给你惊喜。做好监控写好日志留好降级路径。毕竟凌晨三点的告警电话谁都不想接第二次。