如何设置Structured Streaming的分区数？当前分区始终为200

问题根源 #

你遇到的200分区是Spark默认的spark.sql.shuffle.partitions值，而你在readStream的option里设置这个参数是无效的——因为这个参数是全局SQL配置，不属于数据源的可配置选项；numPartitions仅对rate源生效，JSON源不支持该选项。

有效解决方案 #

1. 全局配置shuffle分区数 #

在SparkSession初始化时全局设置，或在代码中动态调整（对后续所有SQL/DSL操作生效）：

// 初始化SparkSession时配置
val spark = SparkSession.builder()
  .appName("StructuredStreamingDemo")
  .config("spark.sql.shuffle.partitions", "6")
  .getOrCreate()

// 代码中动态设置
spark.conf.set("spark.sql.shuffle.partitions", "6")

2. 聚合后手动指定分区数 #

如果不想全局修改配置，可以直接在聚合操作后调用repartition方法指定目标分区数：

val result = jsonDF.groupBy("origin").sum("value").repartition(6)

3. 优化JSON源的初始读取分区（可选） #

若输入JSON文件本身数量多、单文件小，可通过调整以下参数减少初始读取的分区数：

// 调整单分区最大字节数（默认128MB）
spark.conf.set("spark.sql.files.maxPartitionBytes", "128m")
// 限制每次流触发读取的文件数量
val jsonDF = spark.readStream.format("json")
  .schema(schema)
  .option("maxFilesPerTrigger", 10)
  .load("source")

关键说明 #

• spark.sql.shuffle.partitions专门控制shuffle阶段（如groupBy、join）的输出分区数，必须全局设置或在shuffle操作前动态配置，无法作为数据源选项传入readStream。
• 分区数需根据集群CPU核数、内存资源及数据量合理调整，避免过小导致单分区数据过载，或过大产生大量细碎任务拖慢性能。

内容的提问来源于stack exchange，提问作者beatrice