如何在Databricks Spark Streaming foreachBatch中正确启用AQE？

1. 确保参数设置时机正确 #

必须在读取流数据源之前配置AQE参数，不能在流作业启动后设置。正确的代码执行顺序如下：

# 先配置AQE参数
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.forceOptimizeSkewedJoin", "true")

# 再读取流数据并定义流式作业
df = spark.readStream.format("...").load(...)
df.writeStream.format("delta")
        .option("checkpointLocation", "dbfs/loc")
        .foreachBatch(transform_and_upsert)
        .outputMode("update")
        .trigger(availableNow=True)
        .start()

2. 通过集群级配置强制生效 #

会话级配置可能被集群默认配置覆盖，建议直接在集群的Spark配置中添加参数：
在集群编辑页面的「高级选项」->「Spark配置」中添加：

spark.sql.adaptive.enabled true
spark.sql.adaptive.skewJoin.enabled true
spark.sql.adaptive.forceOptimizeSkewedJoin true

保存后重启集群，所有作业都会继承该配置，避免会话级配置被覆盖。

3. 验证参数是否生效 #

在启动流作业前执行以下代码确认参数状态：

print(spark.conf.get("spark.sql.adaptive.enabled"))

若输出true再启动流作业。同时可在Spark UI的「Environment」标签中搜索spark.sql.adaptive.enabled，确认最终生效的配置值。

4. 适配Photon引擎与foreachBatch逻辑 #

• 确保transform_and_upsert函数内使用DataFrame/SQL API，避免直接使用RDD操作（RDD操作不触发AQE优化）。
• 确认Photon引擎已启用（DBR 14.3 Photon集群默认开启，可通过spark.conf.get("spark.databricks.photon.enabled")验证）。

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