Spark withColumn - 使用非“Column”类型变量添加列

Question

我该如何将变量值添加到数据框中的一列？

我知道可以使用.toDF（colName）创建数据框，而.withColumn是添加列的方法。但是，当我尝试以下操作时，会出现类型不匹配错误：

val myList = List(1,2,3)
val myArray = Array(1,2,3)

myList.toDF("myList")
  .withColumn("myArray", myArray)

类型不匹配，期望类型：列(Column)，实际类型：整型数组(Array[Int])

这个编译错误出现在.withColumn方法中的myArray上。如何将它从整型数组(Array[Int])转换为列(Column)类型呢？

- BlueTriangles

Answer 1

错误信息明确指出问题所在，您需要将列（或lit()）作为第二个参数输入到withColumn()中。

请尝试以下操作：

import org.apache.spark.sql.functions.typedLit

val myList = List(1,2,3)
val myArray = Array(1,2,3)

myList.toDF("myList")
  .withColumn("myArray", typedLit(myArray))

:)

- Steven Black

Answer 2

不确定withColumn是否是您实际寻求的内容。您可以应用lit()使myArray符合方法规范，但结果将是DataFrame中每行的相同数组值：

myList.toDF("myList").withColumn("myArray", lit(myArray)).
  show
// +------+---------+
// |myList|  myArray|
// +------+---------+
// |     1|[1, 2, 3]|
// |     2|[1, 2, 3]|
// |     3|[1, 2, 3]|
// +------+---------+

如果您想要按列合并这两个集合，那么这与withColumn提供的不同转换。在这种情况下，您需要将它们各自转换为DataFrame，并通过join组合它们。

现在，如果这两个集合的元素是行标识并且成对地相互匹配（就像您的示例中一样），并且您希望以这种方式加入它们，您可以简单地加入转换后的DataFrames：

myList.toDF("myList").join(
    myArray.toSeq.toDF("myArray"), $"myList" === $"myArray"
  ).show
// +------+-------+
// |myList|myArray|
// +------+-------+
// |     1|      1|
// |     2|      2|
// |     3|      3|
// +------+-------+

但是，如果两个集合中有无法连接的元素，并且您只想将它们按列合并，则需要使用来自两个数据帧的兼容行标识列将它们连接起来。如果没有这样的行标识列，一种方法是创建自己的rowId，如下例所示：

import org.apache.spark.sql.Row
import org.apache.spark.sql.types._

val df1 = List("a", "b", "c").toDF("myList")
val df2 = Array("x", "y", "z").toSeq.toDF("myArray")

val rdd1 = df1.rdd.zipWithIndex.map{
  case (row: Row, id: Long) => Row.fromSeq(row.toSeq :+ id)
}
val df1withId = spark.createDataFrame( rdd1,
  StructType(df1.schema.fields :+ StructField("rowId", LongType, false))
)

val rdd2 = df2.rdd.zipWithIndex.map{
  case (row: Row, id: Long) => Row.fromSeq(row.toSeq :+ id)
}
val df2withId = spark.createDataFrame( rdd2, 
  StructType(df2.schema.fields :+ StructField("rowId", LongType, false))
)

df1withId.join(df2withId, Seq("rowId")).show
// +-----+------+-------+
// |rowId|myList|myArray|
// +-----+------+-------+
// |    0|     a|      x|
// |    1|     b|      y|
// |    2|     c|      z|
// +-----+------+-------+

- Leo C