如何在Spark Structured Streaming中复用已有Schema批量添加空列

原生Spark没有直接支持.withColumns(schema)这种传入StructType直接加列的语法，但可以通过简单的封装实现同类需求，且可以规避手写while循环的潜在问题。

推荐实现（Spark 2.3及以上版本） #

Spark 2.3版本开始withColumns支持传入Map[String, Column]类型的参数，你可以直接把已有schema转换为对应格式实现批量加列，一行代码即可完成：

import org.apache.spark.sql.functions.lit

val controlDataFrame = originalDF.withColumns(
  schema.fields.map(field => field.name -> lit("").cast(field.dataType)).toMap
)

该写法优势：

• 无需手动迭代、不需要声明var变量反复赋值DataFrame
• 自动对齐schema中定义的字段类型，避免后续union时出现类型不兼容报错
• 完全复用已有schema定义，不需要重复编写列名

低版本Spark兼容实现 #

如果你的Spark版本低于2.3，没有带Map参数的withColumns方法，可以用foldLeft函数式写法实现，比while循环更易维护：

import org.apache.spark.sql.functions.lit

val controlDataFrame = schema.fields.foldLeft(originalDF) { (currentDF, field) =>
  currentDF.withColumn(field.name, lit("").cast(field.dataType))
}

原有while写法的优化建议 #

你自己测试的while循环逻辑是可行的，但存在类型不匹配隐患：lit("")默认生成String类型，而你定义的schema中id字段为LongType，直接赋值空字符串会导致列类型和预期schema不一致，后续执行union时会抛出schema不兼容错误，优化后写法如下：

val iter = schema.iterator
while(iter.hasNext) {
  val field = iter.next()
  controlDataFrame = controlDataFrame.withColumn(field.name, lit("").cast(field.dataType))
}

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