从 Spark 数据帧中的单个列派生多个列

Question

我有一个DF，其中包含一个巨大的可解析元数据，作为Dataframe中的单个字符串列，我们称之为DFA，使用ColmnA。

我想通过一个函数，ClassXYZ = Func1(ColmnA)将ColmnA这一列分成多个列。这个函数返回一个具有多个变量的类ClassXYZ，现在每个变量都必须映射到新的列，比如ColmnA1、ColmnA2等。

我如何通过调用Func1一次来完成从一个数据帧到另一个数据帧的转换，而不需要重复它来创建所有的列。

如果我每次都调用这个巨大的函数来添加一个新列，这很容易解决，但这是我希望避免的。

请告知工作或伪代码。

谢谢

桑贾伊

Answer 1

如果结果列的长度与原始列的长度相同，那么可以使用withColumn函数并通过应用udf来创建全新的列。在此之后，您可以删除您的原始专栏，例如:

 val newDf = myDf.withColumn("newCol1", myFun(myDf("originalColumn")))
.withColumn("newCol2", myFun2(myDf("originalColumn"))
.drop(myDf("originalColumn"))

其中myFun是一个定义如下的udf：

   def myFun= udf(
    (originalColumnContent : String) =>  {
      // do something with your original column content and return a new one
    }
  )

Answer 2

假设函数之后有一系列元素，给出如下示例：

val df = sc.parallelize(List(("Mike,1986,Toronto", 30), ("Andre,1980,Ottawa", 36), ("jill,1989,London", 27))).toDF("infoComb", "age")
df.show
+------------------+---+
|          infoComb|age|
+------------------+---+
|Mike,1986,Toronto| 30|
| Andre,1980,Ottawa| 36|
|  jill,1989,London| 27|
+------------------+---+

现在您可以使用此infoComb做的是，您可以开始拆分字符串并获取更多列：

df.select(expr("(split(infoComb, ','))[0]").cast("string").as("name"), expr("(split(infoComb, ','))[1]").cast("integer").as("yearOfBorn"), expr("(split(infoComb, ','))[2]").cast("string").as("city"), $"age").show
+-----+----------+-------+---+
| name|yearOfBorn|   city|age|
+-----+----------+-------+---+
|Mike|      1986|Toronto| 30|
|Andre|      1980| Ottawa| 36|
| jill|      1989| London| 27|
+-----+----------+-------+---+

希望这有帮助。

Answer 3

一般来说，你想要的不是直接可能的。UDF一次只能返回一列。有两种不同的方法可以克服此限制：

>

返回复杂类型的列。最一般的解决方案是结构类型，但您也可以考虑数组类型或映射类型。

import org.apache.spark.sql.functions.udf

val df = Seq(
  (1L, 3.0, "a"), (2L, -1.0, "b"), (3L, 0.0, "c")
).toDF("x", "y", "z")

case class Foobar(foo: Double, bar: Double)

val foobarUdf = udf((x: Long, y: Double, z: String) => 
  Foobar(x * y, z.head.toInt * y))

val df1 = df.withColumn("foobar", foobarUdf($"x", $"y", $"z"))
df1.show
// +---+----+---+------------+
// |  x|   y|  z|      foobar|
// +---+----+---+------------+
// |  1| 3.0|  a| [3.0,291.0]|
// |  2|-1.0|  b|[-2.0,-98.0]|
// |  3| 0.0|  c|   [0.0,0.0]|
// +---+----+---+------------+

df1.printSchema
// root
//  |-- x: long (nullable = false)
//  |-- y: double (nullable = false)
//  |-- z: string (nullable = true)
//  |-- foobar: struct (nullable = true)
//  |    |-- foo: double (nullable = false)
//  |    |-- bar: double (nullable = false)

这可以很容易地变平，但通常没有必要这样做。

转战RDD，重塑重建DF:

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

def foobarFunc(x: Long, y: Double, z: String): Seq[Any] = 
  Seq(x * y, z.head.toInt * y)

val schema = StructType(df.schema.fields ++
  Array(StructField("foo", DoubleType), StructField("bar", DoubleType)))

val rows = df.rdd.map(r => Row.fromSeq(
  r.toSeq ++
  foobarFunc(r.getAs[Long]("x"), r.getAs[Double]("y"), r.getAs[String]("z"))))

val df2 = sqlContext.createDataFrame(rows, schema)

df2.show
// +---+----+---+----+-----+
// |  x|   y|  z| foo|  bar|
// +---+----+---+----+-----+
// |  1| 3.0|  a| 3.0|291.0|
// |  2|-1.0|  b|-2.0|-98.0|
// |  3| 0.0|  c| 0.0|  0.0|
// +---+----+---+----+-----+