Apache Spark Streaming 3.5.3 官方基准测试实战：新手入门与性能调优指南

1次阅读

共计 2445 个字符，预计需要花费 7 分钟才能阅读完成。

实时数据处理面临的核心挑战是如何在有限资源下实现高吞吐与低延迟的平衡。Spark Streaming 采用微批处理（Micro-batch）架构时，开发者常遇到三类典型问题：

资源利用率波动大：突发流量导致 Executor 内存溢出或 CPU 空闲
批次积压（Batch Backlog）：处理速度跟不上数据生成速度
端到端延迟失控：从数据产生到最终输出超过 SLA 要求

Spark 自带的 spark-benchmark 模块包含 StreamingBenchmark 类，其工作原理可概括为：

模拟数据源：生成可配置速率的结构化测试数据
压力注入：通过背压机制（Backpressure）动态调整数据注入速率
度量采集：自动记录各批次处理的延迟、吞吐等 12 项核心指标

Master 节点：m5.2xlarge (8 vCPU, 32GB RAM)
Worker 节点：3 台 r5.xlarge (4 vCPU, 32GB RAM)
存储：EBS gp3 500GB
网络：10Gbps 专用 VPC

# 基础环境
spark-3.5.3-bin-hadoop3.tgz
scala-2.12.18
java-11-openjdk

# Benchmark 依赖
wget https://repo1.maven.org/maven2/org/apache/spark/spark-streaming-benchmark_2.12/3.5.3/spark-streaming-benchmark_2.12-3.5.3.jar

启动 Spark 集群

./sbin/start-master.sh
./sbin/start-worker.sh spark://master:7077 -c 4 -m 8G

提交基准测试作业

spark-submit --class org.apache.spark.sql.streaming.StreamingBenchmark \
  --master spark://master:7077 \
  --executor-memory 6G \
  --total-executor-cores 12 \
  spark-streaming-benchmark_2.12-3.5.3.jar \
  --duration 300 \
  --batch-interval 2000 \
  --rate 100000

关键参数说明：

--duration：测试持续时间（秒）
--batch-interval：批处理间隔（毫秒）
--rate：每秒注入的记录数

间隔(ms)	适用场景	风险点
500	超低延迟场景	可能引发调度开销过高
1000-2000	通用场景	需监控批次积压
≥3000	高吞吐场景	延迟敏感型业务不适用

最优并行度 = min(数据分片数, executor 数 × 每个 executor 核心数 × 2)

实际案例：当 Kafka 主题有 16 个分区时，建议配置：

spark.conf.set("spark.default.parallelism", 16)
spark.conf.set("spark.sql.shuffle.partitions", 16)

指标名称	健康阈值	异常处理建议
处理延迟	≤batchInterval 的 1.5 倍	检查数据倾斜或增大资源
调度延迟	≤50ms	减少小文件数量
吞吐量波动	≤15%	启用动态资源分配

{
  "batchDuration": 2000,
  "processedRowsPerSecond": 98532,
  "avgSchedulingDelay": 23,
  "numActiveReceivers": 2,
  "numRetainedCompletedBatches": 100
}

// 启用动态资源分配
spark.conf.set("spark.dynamicAllocation.enabled", "true")
spark.conf.set("spark.dynamicAllocation.minExecutors", "2")
spark.conf.set("spark.dynamicAllocation.maxExecutors", "10")

// 配置背压机制
spark.conf.set("spark.streaming.backpressure.enabled", "true")
spark.conf.set("spark.streaming.backpressure.initialRate", "50000")

// 识别倾斜键值
val skewedKeys = spark.table("input_stream")
  .groupBy("user_id")
  .count()
  .filter("count > 100000") // 阈值根据业务调整

// 倾斜处理方案
val processed = rawStream.transform { rdd =>
  rdd.mapPartitions { iter =>
    if (TaskContext.get.partitionId == 0) {
      // 特殊处理倾斜分区
      iter.flatMap(handleSkewedRecords)
    } else {iter}
  }
}