scala - 如何将 List[Double] 转换为 Columns？

Question

我有 List[Double] ，如何将其转换为 org.apache.spark.sql.Column .我正在尝试使用 .withColumn() 将其作为列插入到现有 DataFrame .

Answer 1

不能直接完成。 Column不是数据结构，而是特定 SQL 表达式的表示。它不受特定数据的约束。您必须首先转换数据。解决此问题的一种方法是 parallelize和 join按索引:

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

val df = Seq(("a", 2), ("b", 1), ("c", 0)).toDF("x", "y")
val aList = List(1.0, -1.0, 0.0)

val rows = df.rdd.zipWithIndex.map(_.swap)
  .join(sc.parallelize(aList).zipWithIndex.map(_.swap))
  .values
  .map { case (row: Row, x: Double) => Row.fromSeq(row.toSeq :+ x) }

sqlContext.createDataFrame(rows, df.schema.add("z", DoubleType, false))

另一种类似的方法是索引和使用和 UDF 来处理其余的:

import scala.util.Try

val indexedDf = sqlContext.createDataFrame(
  df.rdd.zipWithIndex.map {
    case (row: Row, i: Long) => Row.fromSeq(row.toSeq :+ i)
  },
  df.schema.add("idx_", "long")
)

def addValue(vs: Vector[Double]) = udf((i: Long) => Try(vs(i.toInt)).toOption)

indexedDf.withColumn("z", addValue(aList.toVector)($"idx_"))

不幸的是，这两种解决方案都会受到这些问题的影响。首先，通过驱动程序传递本地数据会在您的程序中引入严重的瓶颈。通常数据应该直接从执行程序访问。如果你想迭代地执行这个操作，另一个问题是不断增长的 RDD 谱系。

虽然第二个问题可以通过检查点来解决，但第一个问题使这个想法通常毫无用处。我强烈建议您首先构建完整的结构，然后在 Spark 上阅读它，或者以可以利用 Spark 架构的方式重建管道。例如，如果数据来自外部源，则使用 map 直接对每个数据块执行读取操作。/ mapPartitions .