Spark Structured Streaming中为每个StreamingQuery设置不同的spark.sql.shuffle.partitions

在同一个SparkSession中给不同StreamingQuery设置独立的spark.sql.shuffle.partitions，可以通过以下两种方案实现，结合你的多查询共用Session、独立检查点的场景，优先推荐第一种轻量方案：

方案1：在查询数据流中局部临时设置参数 #

利用Dataset.transform()方法，在每个查询的处理链路中临时修改配置，该配置仅对当前查询的执行计划生效，不会影响其他查询。因为每个StreamingQuery的作业是独立提交的，Spark会在作业执行时读取当前作用域内的配置。

示例代码（Scala）：

// 全局默认配置（可选，不设置则用Spark默认值）
spark.conf.set("spark.sql.shuffle.partitions", "200")

// 第一个查询：设置shuffle分区为100
val query1 = sourceDF1
  .transform { ds =>
    // 临时修改当前查询的shuffle分区配置
    ds.sparkSession.conf.set("spark.sql.shuffle.partitions", "100")
    ds
  }
  .groupBy($"user_id")
  .agg(count($"event") as "event_count")
  .writeStream
  .format("parquet")
  .option("checkpointLocation", "/data/checkpoints/query1")
  .start()

// 第二个查询：设置shuffle分区为300
val query2 = sourceDF2
  .transform { ds =>
    ds.sparkSession.conf.set("spark.sql.shuffle.partitions", "300")
    ds
  }
  .join(dimDF, $"item_id" === dimDF.id)
  .writeStream
  .format("delta")
  .option("checkpointLocation", "/data/checkpoints/query2")
  .start()

// 等待任意查询终止
spark.streams.awaitAnyTermination()

关键注意事项 #

• 必须在触发Shuffle的算子（如groupBy、join、distinct）之前执行conf.set，否则配置不会对Shuffle操作生效。
• 修改该参数后，一定要清理对应查询的检查点路径，否则检查点中保存的旧Shuffle状态会和新配置冲突，导致查询启动失败。

方案2：为每个查询创建独立子SparkSession #

如果需要更彻底的配置隔离，可以基于主SparkSession创建子Session，子Session会继承主Session的配置，但可以单独修改参数，每个查询使用自己的子Session。不过这种方式会增加资源开销，因为每个子Session有独立的Catalog和执行上下文，适合资源充足的场景。

示例代码（Scala）：

// 主SparkSession
val spark = SparkSession.builder()
  .appName("MultiStreamingQuery")
  .getOrCreate()

// 子Session1：配置shuffle分区为100
val spark1 = spark.newSession()
spark1.conf.set("spark.sql.shuffle.partitions", "100")
val query1 = spark1.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker:9092")
  .option("subscribe", "topic1")
  .load()
  .selectExpr("CAST(value AS STRING)")
  .writeStream
  .option("checkpointLocation", "/data/checkpoints/query1")
  .format("parquet")
  .start()

// 子Session2：配置shuffle分区为300
val spark2 = spark.newSession()
spark2.conf.set("spark.sql.shuffle.partitions", "300")
val query2 = spark2.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker:9092")
  .option("subscribe", "topic2")
  .load()
  .groupBy(window($"timestamp", "10 minutes"))
  .agg(count("*") as "window_count")
  .writeStream
  .option("checkpointLocation", "/data/checkpoints/query2")
  .format("parquet")
  .start()

// 监听查询终止，停止所有查询
val allQueries = Seq(query1, query2)
spark.streams.addListener(new StreamingQueryListener {
  override def onQueryStarted(event: QueryStartedEvent): Unit = {}
  override def onQueryProgress(event: QueryProgressEvent): Unit = {}
  override def onQueryTerminated(event: QueryTerminatedEvent): Unit = {
    allQueries.foreach(_.stop())
  }
})

spark.streams.awaitAnyTermination()

关键注意事项 #

• 子Session的检查点路径必须完全独立，避免和其他Session的查询冲突。
• 子Session会占用额外的内存和资源，需要根据集群资源情况评估是否适用。

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