【问题标题】:rename spark dataframe structType fields重命名火花数据帧结构类型字段
【发布时间】:2020-08-26 01:51:39
【问题描述】:

给定一个动态的 structType 。这里 structType 名称未知。它是动态的,因此它的名称正在改变。

名称是可变的。所以不要在架构中预先假设“MAIN_COL”。

root
 |-- MAIN_COL: struct (nullable = true)
 |    |-- a: string (nullable = true)
 |    |-- b: string (nullable = true)
 |    |-- c: string (nullable = true)
 |    |-- d: string (nullable = true)
 |    |-- f: long (nullable = true)
 |    |-- g: long (nullable = true)
 |    |-- h: long (nullable = true)
 |    |-- j: long (nullable = true)

我们如何编写动态代码以重命名 structType 的字段,并将其名称作为前缀。

root
 |-- MAIN_COL: struct (nullable = true)
 |    |-- MAIN_COL_a: string (nullable = true)
 |    |-- MAIN_COL_b: string (nullable = true)
 |    |-- MAIN_COL_c: string (nullable = true)
 |    |-- MAIN_COL_d: string (nullable = true)
 |    |-- MAIN_COL_f: long (nullable = true)
 |    |-- MAIN_COL_g: long (nullable = true)
 |    |-- MAIN_COL_h: long (nullable = true)
 |    |-- MAIN_COL_j: long (nullable = true)

【问题讨论】:

  • 一种方法是,解压嵌套列并根据需要重命名

标签: scala apache-spark apache-spark-sql spark-streaming


【解决方案1】:

您可以使用 DSL 更新嵌套列的架构。

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

val schema: StructType = df.schema.fields.head.dataType.asInstanceOf[StructType]

val updatedSchema = StructType.apply(
       schema.fields.map(sf => StructField.apply("MAIN_COL_" + sf.name, sf.dataType))
)

val resultDF = df.withColumn("MAIN_COL", $"MAIN_COL".cast(updatedSchema))

更新架构:

root
 |-- MAIN_COL: struct (nullable = false)
 |    |-- MAIN_COL_a: string (nullable = true)
 |    |-- MAIN_COL_b: string (nullable = true)
 |    |-- MAIN_COL_c: string (nullable = true)

【讨论】:

  • 名称是变量。所以不要在架构中预先假设“MAIN_COL”。
猜你喜欢
  • 1970-01-01
  • 2017-08-17
  • 1970-01-01
  • 2016-05-06
  • 2021-07-08
  • 2018-11-08
  • 1970-01-01
  • 2020-08-09
  • 2018-02-15
相关资源
最近更新 更多