Spark SQL对比多列CSV遇Max iterations报错的原因与优化咨询

问题背景 #

我在Databricks上使用Spark 2.4.4 + Java做两个大型CSV的差异对比，需要识别：

• 仅基于Id的新增记录（插入行）
• 全列对比的修改行（排除新增行）
• 仅基于Id的删除行

实现代码如下：

Dataset<Row> insertedDf = newDf_temp.join(oldDf_temp,oldDf_temp.col(key) .equalTo(newDf_temp.col(key)),"left_anti");
Long insertedCount = insertedDf.count();
logger.info("Inserted File Count == "+insertedCount);

Dataset<Row> deletedDf = oldDf_temp.join(newDf_temp,oldDf_temp.col(key) .equalTo(newDf_temp.col(key)),"left_anti") .select(oldDf_temp.col(key));
Long deletedCount = deletedDf.count();
logger.info("deleted File Count == "+deletedCount);

Dataset<Row> changedDf = newDf_temp.exceptAll(oldDf_temp); // 得到新增+修改记录
Dataset<Row> changedDfTemp = changedDf.join(insertedDf, changedDf.col(key) .equalTo(insertedDf.col(key)),"left_anti"); // 仅得到修改记录
Long changedCount = changedDfTemp.count();
logger.info("Changed File Count == "+changedCount);

在列数≤50的CSV中运行正常，但300+列时触发异常：Max iterations (100) reached for batch Resolution – Spark Error，设置sparkConf.set("spark.sql.optimizer.maxIterations", "500")后恢复正常。

为什么需要设置spark.sql.optimizer.maxIterations参数？ #

Spark SQL的优化器在生成查询计划时，会通过多轮迭代应用各种优化规则（比如谓词下推、列裁剪、Join逻辑重排等）。当处理300+列的CSV时，查询计划的复杂度会陡增：

• exceptAll操作需要对比所有列的内容，会生成包含数百列的比较逻辑
• 后续的Join操作又会叠加更多列关联逻辑，优化器每一轮迭代都要处理大量列相关的规则校验

默认的100次迭代不足以让优化器完成所有必要的规则应用，因此触发了“达到最大迭代次数”的异常。调高这个参数，就是给优化器足够的时间完成全部优化步骤。

你的操作是否有误？ #

从业务逻辑层面来说，你的增删改识别逻辑是完全正确的：

• 用left_anti Join查找新增/删除记录是高效且标准的实现方式
• 用exceptAll结合left_anti过滤修改记录的逻辑，也能准确区分新增和修改的差异

问题出在多列场景下查询计划的复杂度上，而非你的业务逻辑本身。

这是多列CSV的预期行为吗？ #

在Spark 2.x（尤其是2.4.x）版本中，这属于常见的边缘场景表现。早期Spark SQL的优化器在处理超大量列的复杂查询时，迭代效率不如后续的Spark 3.x版本。当列数达到数百级别时，每一轮优化需要处理的规则数量和计划复杂度都会大幅上升，导致默认迭代次数不够用。这可以看作是Spark 2.4.x处理多列数据时的一个已知限制，而非Bug。

其他优化方法处理多列CSV？ #

除了调高迭代次数，还有这些更高效的优化方案：

1. 提前裁剪不必要的列
   如果部分列不需要参与对比（比如日志字段、冗余标识列），读取CSV时就只加载需要的列，直接降低查询计划的复杂度：

// 定义需要参与对比的列列表
List<String> neededColumns = Arrays.asList("Id", "col1", "col2", ...);
Dataset<Row> newDf_temp = spark.read()
    .option("header", "true")
    .csv(newPath)
    .select(neededColumns.stream().map(functions::col).toArray(Column[]::new));

2. 替换exceptAll为哈希对比逻辑
   exceptAll会自动对比所有列，你可以将需要对比的列拼接成字符串并生成哈希值，把多列对比转化为单哈希列对比，大幅简化查询计划：

// 定义所有需要对比的列（排除Id）
List<Column> compareColumns = Arrays.asList(col("col1"), col("col2"), ...);
// 生成数据哈希（注意处理空值，避免拼接异常）
Dataset<Row> newDfWithHash = newDf_temp.withColumn("data_hash", 
    functions.sha2(functions.concat_ws("|", compareColumns), 256));
Dataset<Row> oldDfWithHash = oldDf_temp.withColumn("data_hash", 
    functions.sha2(functions.concat_ws("|", compareColumns), 256));

// 通过Id关联后对比哈希值，筛选修改记录
Dataset<Row> changedDfTemp = newDfWithHash.join(oldDfWithHash, 
    newDfWithHash.col(key).equalTo(oldDfWithHash.col(key)), "inner")
    .where(newDfWithHash.col("data_hash").notEqual(oldDfWithHash.col("data_hash")))
    .select(newDfWithHash.columns());

3. 升级到Spark 3.x版本
   Spark 3.x对SQL优化器做了大量底层改进，包括优化规则的执行效率、迭代逻辑的优化，在处理多列数据时的表现远优于2.4.x版本。如果Databricks环境允许升级，这是最彻底的解决方案。

4. 调整优化器批次参数
   除了maxIterations，还可以尝试调整spark.sql.optimizer.maxIterationsPerBatch参数，控制每个优化批次的迭代次数，但这个参数在2.4.x中生效有限，不如直接调高总迭代次数来得直接。

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