时间戳

首页 > 解决方案 > Spark - 使用 df.schema.copy 函数为另一个数据帧复制一个字段

scala - Spark - 使用 df.schema.copy 函数为另一个数据帧复制一个字段

问题描述

我需要使用现有的 df 字段创建一个模式。

考虑这个示例数据框

scala> case class prd (a:Int, b:Int)
defined class prd

scala> val df = Seq((Array(prd(10,20),prd(15,30),prd(20,25)))).toDF("items")
df: org.apache.spark.sql.DataFrame = [items: array<struct<a:int,b:int>>]

scala> df.printSchema
root
 |-- items: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- a: integer (nullable = false)
 |    |    |-- b: integer (nullable = false)

我还需要一个类似于 df2 的“items”的字段“items_day1”。现在，我正在像下面那样做，这是一种解决方法 

scala> val df2=df.select('items,'items.as("item_day1"))
df2: org.apache.spark.sql.DataFrame = [items: array<struct<a:int,b:int>>, item_day1: array<struct<a:int,b:int>>]

scala> df2.printSchema
root
 |-- items: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- a: integer (nullable = false)
 |    |    |-- b: integer (nullable = false)
 |-- item_day1: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- a: integer (nullable = false)
 |    |    |-- b: integer (nullable = false)


scala>

但是如何使用 df.schema.add() 或 df.schema.copy() 函数来获得它？

编辑1：

我正在尝试如下

val (a,b) = (df.schema,df.schema) // works
a("items")  //works
b.add(a("items").as("items_day1")) //Error..

标签： scalaapache-sparkapache-spark-sql

解决方案

要将具有相同结构但与现有字段的顶级名称不同的新字段添加到 DataFrame 模式（属于StructType ），您可以使用修改后的 StructField 成员复制StructFieldname，如下所示：

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

case class prd (a:Int, b:Int)

val df = Seq((Array(prd(10,20), prd(15,30), prd(20,25)))).toDF("items")

val schema = df.schema
// schema: org.apache.spark.sql.types.StructType = StructType(
//   StructField(items, ArrayType(
//     StructType(StructField(a,IntegerType,false), StructField(b,IntegerType,false)
//   ), true), true)
// )

val newSchema = schema.find(_.name == "items") match {
  case Some(field) => schema.add(field.copy(name = "items_day1"))
  case None        => schema
}
// newSchema: org.apache.spark.sql.types.StructType = StructType(
//   StructField(items, ArrayType(
//     StructType(StructField(a,IntegerType,false), StructField(b,IntegerType,false)
//   ), true), true),
//   StructField(items_day1, ArrayType(
//     StructType(StructField(a,IntegerType,false), StructField(b,IntegerType,false)
//   ), true), true)
// )

推荐阅读