Spark 在 UDAF 之后将 SQL Row 展平为新列答案

【问题标题】：Spark flatten SQL Row into new columns after UDAFSpark 在 UDAF 之后将 SQL Row 展平为新列
【发布时间】：2016-08-05 06:13:31
【问题描述】：

我有一个 Spark 数据框，我在其上执行 groupBy、用户定义的聚合和库聚合：

data.groupBy("var1").agg(sum("var2"), makefreqs("var3")).first

产生类似的东西：

org.apache.spark.sql.Row = ["var1_value1", 219, WrappedArray(0.6, 0.1, 0.3)]

我最终想做的是将所有行（不仅仅是 .first）转换成这种格式：

org.apache.spark.mllib.linalg.Vector = [219, 0.6, 0.1, 0.3]

我不知道如何在 sql.Row 中将 WrappedArray(0.6, 0.1, 0.3) 展平为三个新列（最好带有名称）。最聪明的方法是什么？

【问题讨论】：

标签： scala apache-spark apache-spark-sql

【解决方案1】：

您可以在Apache DataFu 中找到所需的开箱即用解决方案。 explodeArray 方法完全符合您的需要：

import datafu.spark.DataFrameOps._

val df = sc.parallelize(Seq(("var1_value1",219,Array(0.6, 0.1, 0.3)))).toDF
df.explodeArray(col("_3"), "array").show

这将产生：

+-----------+---+---------------+------+------+------+
|    _1     | _2|       _3      |array0|array1|array2|
+-----------+---+---------------+------+------+------+
|var1_value1|219|[0.6, 0.1, 0.3]|   0.6|   0.1|   0.3|
+-----------+---+---------------+------+------+------+

请考虑，为了评估要为数组创建多少列，此方法会评估数据框 - 如果计算成本很高，则应将其缓存。

【讨论】：

【解决方案2】：

我不确定最聪明的方法，但这些是我的建议。

一种方法，你可以使用 map 函数，但它会变成一个 RDD，所以你需要将它转换回 DataFrame。

val df = data.groupBy("var1").agg(sum("var2"), makefreqs("var3"))

// implicitly pass context to toDF
import sqlContext.implicits._

df.map { case Row(var1, sumVar2, array: WrappedArray) => 
  Row(var1, sumVar2, array(0), array(1), array(2) }
  .toDF("var1", "sum_var2", "a1", "a2", "a3")

如果你不喜欢来回转换，你可以使用 udf 来完成任务。

val arrayToColumn = (index: Int) => udf { (array: Seq[Double]) => array(index) }

df
  .withColumn("a0", arrayToColumn(0)(df("array_col")))
  .withColumn("a1", arrayToColumn(1)(df("array_col")))
  .withColumn("a2", arrayToColumn(2)(df("array_col")))

【讨论】：