scala - 如何在不从 DataFrame 转换并访问它的情况下向 Dataset 添加列？

Question

我知道使用 .withColumn() 向 Spark DataSet 添加新列的方法和一个 UDF ，它返回一个数据帧。我也知道，我们可以将生成的 DataFrame 转换为 DataSet。

我的问题是:

如果我们仍然遵循传统的 DF 方法(即将列名作为 UDF 输入的字符串传递)，DataSet 的类型安全如何在这里发挥作用

是否有一种“面向对象的方式”来访问列(不将列名作为字符串传递)，就像我们过去对 RDD 所做的那样，用于附加新列。

如何在 map 、过滤器等正常操作中访问新列？

例如:

    scala> case class Temp(a : Int, b : String)    //creating case class
    scala> val df = Seq((1,"1str"),(2,"2str),(3,"3str")).toDS    // creating DS
    scala> val appendUDF = udf( (b : String) => b + "ing")      // sample UDF

    scala> df.withColumn("c",df("b"))   // adding a new column
    res5: org.apache.spark.sql.DataFrame = [a: int, b: string ... 1 more field]

    scala> res5.as[Temp]   // converting to DS
    res6: org.apache.spark.sql.Dataset[Temp] = [a: int, b: string ... 1 more field]

    scala> res6.map( x =>x.  
    // list of autosuggestion :
    a   canEqual   equals     productArity     productIterator   toString   
    b   copy       hashCode   productElement   productPrefix 

新栏目 c ，我使用 .withColumn() 添加的不可访问，因为列 c不在案例类 Temp (它仅包含 a 和 b )在使用 res5.as[Temp] 转换为 DS 的瞬间.

如何访问专栏 c ?

Answer 1

在 Dataset 的类型安全世界中你会将一个结构映射到另一个结构中。

也就是说，对于每个转换，我们都需要数据的模式表示(因为它是 RDD 所需要的)。要访问上面的“c”，我们需要创建一个新模式来提供对它的访问。

case class A(a:String)
case class BC(b:String, c:String)
val f:A => BC = a=> BC(a.a,"c") // Transforms an A into a BC

val data = (1 to 10).map(i => A(i.toString))
val dsa = spark.createDataset(data)
// dsa: org.apache.spark.sql.Dataset[A] = [a: string]

val dsb = dsa.map(f)
//dsb: org.apache.spark.sql.Dataset[BC] = [b: string, c: string]