Spark集群任务报错：Union仅支持同列数表，求排查解决

错误原因分析 #

从报错的执行计划可以定位两个核心问题：

1. 上游数据集冗余列未清理：inputEntries在生成过程中（通过窗口函数计算isDuplicate后），没有彻底移除临时生成的row_number列，该冗余列被带入后续Join流程。虽然你在最后select了目标列，但集群模式下Spark的执行优化会保留更多中间列以提升性能，导致冗余列未被及时清理，最终在内部隐式Union操作（如分区合并、优化步骤触发）时，出现7列与8列的不匹配。
2. 无效列操作干扰逻辑：代码中drop("otherDataset.isDuplicate")是无效操作——因为otherEntries已经通过select(joinKey)只保留了rowId列，根本不存在otherDataset.isDuplicate。这行冗余代码可能干扰Spark的列解析逻辑，间接导致列数异常。

修复步骤 #

1. 清理上游数据集的冗余列 #

在生成inputEntries的窗口计算步骤后，显式选择ReconEntity需要的7列，彻底丢弃row_number等临时列：

// 示例：修改inputEntries生成逻辑，确保只保留目标列
val inputEntries = originalData
  .withColumn("row_number", row_number().over(Window.partitionBy("rowId").orderBy(col("processingDate").desc)))
  .withColumn("isDuplicate", when(col("row_number") === 1, "false").otherwise("true"))
  // 显式指定需要的列，移除临时列
  .select("rowId", "groupId", "amounts", "processingDate", "attributes", "entityType", "isDuplicate")
  .as[ReconEntity]

2. 优化Join后的代码逻辑 #

移除无效的drop操作，直接计算最终的isDuplicate，同时明确指定目标列，避免Spark保留多余列：

val result = inputEntries.as("inputDataset")
  .join(
    otherEntries.as("otherDataset").select(joinKey),
    Seq(joinKey),
    "left"
  )
  // 直接生成最终的isDuplicate列，无需中间列
  .withColumn("isDuplicate", when(col("otherDataset.rowId").isNull, "true").otherwise(col("inputDataset.isDuplicate")))
  // 显式选择目标列，明确列来源
  .select(
    col("inputDataset.rowId"),
    col("inputDataset.groupId"),
    col("inputDataset.amounts"),
    col("inputDataset.processingDate"),
    col("inputDataset.attributes"),
    col("inputDataset.entityType"),
    col("isDuplicate")
  )
  .as[ReconEntity]

3. 验证列结构一致性 #

在关键步骤后添加printSchema()，确保每一步的数据集列数和结构完全符合ReconEntity的定义：

inputEntries.printSchema() // 确认输出为7列
otherEntries.select(joinKey).printSchema() // 确认输出为1列
result.printSchema() // 确认输出为7列

补充说明 #

本地运行正常是因为本地Spark的优化策略更激进，会及时清理冗余列；而集群模式下Spark会保留更多中间列以优化分布式执行，导致冗余列被带入后续操作引发Union错误。

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