Spark Streaming 反压机制原理剖析:从控制论到生产调优实战指南

Kafka Topic (持续流入)
   ↓
[DStream]  ← 按 batch interval 切片 (如 1s 一批)
   ↓
[Receiver / DirectKafka]  ← 从 Kafka 拉数据
   ↓
[RDD DAG]  ← 每批生成一个 RDD
   ↓
[TaskScheduler]  ← 调度任务到 Executor
   ↓
[Executors]  ← 执行计算
   ↓
[结果写回]  →  HDFS / Kafka / DB

时间轴 ─────────────────────────────────────────→
批1   批2   批3   批4   批5   批6   批7
└─1s──└─1s──└─1s──└─1s──└─1s──└─1s──└─1s──  (Batch Interval)
实际处理 200ms 200ms 200ms 200ms 200ms 200ms 200ms
↑
正常状态：处理远快于间隔
突然：
批1   批2   批3   批4   批5   批6
└─1s──└─1s──└─1s──└─1s──└─1s──└─1s──
处理 1.5s 1.5s 1.5s 1.5s 1.5s 1.5s
↑
积压！Scheduling Delay 越来越高 → Executor OOM → 任务失败

        ┌──────────────────────────────────────┐
        │           PID Controller              │
        │  ┌──────┐  ┌──────┐  ┌──────┐         │
误差 ──→│  │  P   │  │  I   │  │  D   │  ──→ 输出速率 │
        │  └──────┘  └──────┘  └──────┘         │
        │       (比例)  (积分)  (微分)            │
        └──────────────────────────────────────┘

error(t) = processing_time(t) - batch_interval
            ↑ 实际处理时间    ↑ 期望时间
rate(t+1) = rate(t) - integral_error - rate_error - proportional_error
              ↑ 旧速率      ↑ I 项        ↑ D 项        ↑ P 项

处理时间 < Batch Interval → 速度过慢 → error 负 → 提高 rate
处理时间 > Batch Interval → 速度过快 → error 正 → 降低 rate
处理时间 = Batch Interval → 平衡状态 → rate 保持

spark.streaming.kafka.maxRatePerPartition    // Kafka 每分区最大速率
spark.streaming.backpressure.enabled        // 启用反压（已弃用）
spark.streaming.receiver.writeAheadLog.enable  // WAL

spark.streaming.backpressure.enabled = true
spark.streaming.kafka.maxRatePerPartition =  // 上限保护

// spark/streaming/scheduler/rate/PIDRateController.scala
class PIDRateController(
    conf: SparkConf,
    estimator: RateEstimator
) extends RateController(conf, estimator) {
  def compute(
      time: Long,       // 当前批时间戳
      elements: Long,   // 上一批元素数
      processingDelay: Long,  // 上一批处理耗时
      schedulingDelay: Long   // 上一批调度延迟
  ): Option[Double] = {
    // 误差 = 处理延迟 + 调度延迟
    val error = schedulingDelay.toDouble / 1000  // 转为秒
    val rate = ...
    if (error > 0) {
      // 积压了，需要降速
      newRate = oldRate * (1 - error / proportional)
    } else {
      // 没积压，可以提一点速
      newRate = oldRate * (1 - integralError / integral)
    }
    Some(newRate)
  }
}

// spark 源码
val proportional = conf.getTimeAsMs("spark.streaming.backpressure.proportional", "1s")
val integral = conf.getTimeAsMs("spark.streaming.backpressure.integral", "0.5s")
val derivative = conf.getTimeAsMs("spark.streaming.backpressure.derivative", "0")
val minRate = conf.getDouble("spark.streaming.backpressure.pid.minRate", 100)
newRate = oldRate - proportionalTerm - integralTerm - derivativeTerm

val spark = SparkSession.builder
  .appName("BackpressureDemo")
  .config("spark.streaming.backpressure.enabled", "true")
  .getOrCreate()
val ssc = new StreamingContext(spark.sparkContext, Seconds(2))
ssc.sparkContext.setLogLevel("WARN")
// 启用反压（推荐显式设置）
spark.conf.set("spark.streaming.kafka.maxRatePerPartition", "10000")

spark-submit \
  --conf spark.streaming.backpressure.enabled=true \
  --conf spark.streaming.kafka.maxRatePerPartition=10000 \
  --class com.example.MyApp \
  my-app.jar

# 基础流配置
spark.streaming.backpressure.enabled       true
spark.streaming.kafka.maxRatePerPartition  10000  # 上限保护
# 内存配置（反压后内存压力小，可适当调大）
spark.executor.memory                      4g
spark.executor.memoryOverhead              1g
spark.streaming.backpressure.initialRate  5000   # 初始速率
# 序列化（推荐 Kryo）
spark.serializer                           org.apache.spark.serializer.KryoSerializer

1. 启用反压 → 观察 Scheduling Delay
                ↓ Scheduling Delay 接近 0
2. 调小 Batch Interval（如 1s→500ms）提高实时性
                ↓
3. 提升 Kafka maxRatePerPartition（但设上限保护）
                ↓
4. 调大并行度（增加 Kafka topic 分区数 + Executor 数）
                ↓
5. 优化单批处理逻辑（去除 shuffle、用 foreachPartition 代替 foreach）

// ❌ Receiver 模式
val kafkaStream = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap)
// ✅ Direct 模式
val directStream = KafkaUtils.createDirectStream[String, String](
  ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
)

上游 Task A  →  [Netty Buffer]  →  下游 Task B
   信用 = 8     缓冲: 5 / 8        消费 3 个
                ↑                  ↓
                └── 给 A 发 Credit: 还剩 5 个
                          ↓
                A 最多发 5 个（不超 buffer 上限）

val df = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host:9092")
  .option("subscribe", "topic")
  .load()
df.writeStream
  .format("console")
  .option("checkpointLocation", "/path")
  .start()
  .awaitTermination()

原理：PID 控制器（P比例 + I积分 + D微分）动态调整 Kafka 拉取速率
目标：processing_time ≈ batch_interval，scheduling_delay ≈ 0
配置：spark.streaming.backpressure.enabled = true
上限：spark.streaming.kafka.maxRatePerPartition = 10000
参数：proportional=1s, integral=0.5s, minRate=100
监控：Spark UI → Streaming → Scheduling Delay / Processing Time
对比：Spark PID（秒级） vs Flink Credit（毫秒级）
演进：DStream → Structured Streaming（continuous mode）