如何将数组元素作为新列逐行添加到Spark DataFrame（Scala/Java实现）

Scala 实现方案 #

方案1：行索引关联法（兼容性最优，推荐） #

通过给两份数据分别添加连续行索引后关联，可适配绝大多数场景：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window

// 给原employees DataFrame 添加从0开始的连续行索引
// 若要保证行顺序稳定，可将orderBy参数替换为确定的排序字段，如orderBy("name")
val employeesWithIdx = employees.withColumn("row_idx", row_number().over(Window.orderBy(lit(1))) - 1)

// 将state数组转换为带行索引的DataFrame
val stateArr = Array("LA", "AZ", "OH")
val stateDF = spark.createDataFrame(stateArr.zipWithIndex).toDF("state", "row_idx")

// 关联后删除索引列得到最终结果
val resDF = employeesWithIdx.join(stateDF, Seq("row_idx")).drop("row_idx")

// 输出验证
resDF.show()

方案2：RDD zip 法（代码更简洁） #

当确认employees行数和state数组长度完全一致时，可以直接通过RDD的zip操作拼接数据：

import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{StringType, StructField, StructType}

val stateArr = Array("LA", "AZ", "OH")
// 生成和原DF分区数一致的state RDD
val stateRDD = spark.sparkContext.parallelize(stateArr, employees.rdd.getNumPartitions)

// 拼接两行数据
val combinedRDD = employees.rdd.zip(stateRDD).map {
  case (originRow, stateVal) => Row.fromSeq(originRow.toSeq :+ stateVal)
}

// 构建新的Schema
val newSchema = StructType(employees.schema.fields :+ StructField("state", StringType))

// 转回DataFrame
val resDF = spark.createDataFrame(combinedRDD, newSchema)

// 输出验证
resDF.show()

注意：Spark DataFrame本身无内置行顺序的定义，两种方案都需要你确认原DF的行顺序符合预期，若对顺序稳定性要求高，建议给原DF添加明确的排序规则。

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