scala – 如何将List [Double]转换为列？

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直接读取每个数据块.