重命名Spark DataFrame中嵌套结构的列。

Question

重命名Spark DataFrame中嵌套结构的列。

9

我正在尝试在Scala中更改DataFrame列的名称。我可以轻松地更改直接字段的列名，但在转换数组结构列时遇到了困难。

以下是我的DataFrame模式。

|-- _VkjLmnVop: string (nullable = true)
|-- _KaTasLop: string (nullable = true)
|-- AbcDef: struct (nullable = true)
 |    |-- UvwXyz: struct (nullable = true)
 |    |    |-- _MnoPqrstUv: string (nullable = true)
 |    |    |-- _ManDevyIxyz: string (nullable = true)

但我需要以下这样的模式(schema)。

|-- vkj_lmn_vop: string (nullable = true)
|-- ka_tas_lop: string (nullable = true)
|-- abc_def: struct (nullable = true)
 |    |-- uvw_xyz: struct (nullable = true)
 |    |    |-- mno_pqrst_uv: string (nullable = true)
 |    |    |-- man_devy_ixyz: string (nullable = true)

对于非结构化的列，我通过以下方式更改列名：

def aliasAllColumns(df: DataFrame): DataFrame = {
  df.select(df.columns.map { c =>
    df.col(c)
      .as(
        c.replaceAll("_", "")
          .replaceAll("([A-Z])", "_$1")
          .toLowerCase
          .replaceFirst("_", ""))
  }: _*)
}
aliasAllColumns(file_data_df).show(1)

我该如何动态更改结构体列名？

- Vijay

你是否有重命名列的功能，例如将 Maps(_VkjLmnVop => vkj_lmn_vop, _KaTasLop => ka_tas_lop)？ - stack0114106

@stack0114106，我有很多列。所以我想动态地更改列名。 - Vijay

2个回答

0

据我所知，直接重命名嵌套字段是不可行的。

一方面，您可以尝试移动到平面对象。

但是，如果您需要保留结构，可以使用spark.sql.functions.struct(*cols)进行尝试。

Creates a new struct column.
Parameters: cols – list of column names (string) or list of Column expressions

您需要将所有模式分解，生成所需的别名，然后使用struct函数重新组合。

这不是最好的解决方案。但总比没有强 :)

Pd：我附上了PySpark文档，因为它包含比Scala更好的解释。

- Franzi

但是我有10个struct列，每个结构体有18个属性？有更好的方法吗？ - Vijay

我建议您编写一个尾递归函数，该函数接收一个模式，并生成所有的替换/结构方法。 - Franzi

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- Leo C · Accepted Answer

您可以创建一个递归方法来遍历DataFrame模式以重命名列：

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

def renameAllCols(schema: StructType, rename: String => String): StructType = {
  def recurRename(schema: StructType): Seq[StructField] = schema.fields.map{
      case StructField(name, dtype: StructType, nullable, meta) =>
        StructField(rename(name), StructType(recurRename(dtype)), nullable, meta)
      case StructField(name, dtype: ArrayType, nullable, meta) if dtype.elementType.isInstanceOf[StructType] =>
        StructField(rename(name), ArrayType(StructType(recurRename(dtype.elementType.asInstanceOf[StructType])), true), nullable, meta)
      case StructField(name, dtype, nullable, meta) =>
        StructField(rename(name), dtype, nullable, meta)
    }
  StructType(recurRename(schema))
}

使用以下示例进行测试：

import org.apache.spark.sql.functions._
import spark.implicits._

val renameFcn = (s: String) =>
  s.replace("_", "").replaceAll("([A-Z])", "_$1").toLowerCase.dropWhile(_ == '_')

case class C(A_Bc: Int, D_Ef: Int)

val df = Seq(
  (10, "a", C(1, 2), Seq(C(11, 12), C(13, 14)), Seq(101, 102)),
  (20, "b", C(3, 4), Seq(C(15, 16)), Seq(103))
).toDF("_VkjLmnVop", "_KaTasLop", "AbcDef", "ArrStruct", "ArrInt")

val newDF = spark.createDataFrame(df.rdd, renameAllCols(df.schema, renameFcn))

newDF.printSchema
// root
//  |-- vkj_lmn_vop: integer (nullable = false)
//  |-- ka_tas_lop: string (nullable = true)
//  |-- abc_def: struct (nullable = true)
//  |    |-- a_bc: integer (nullable = false)
//  |    |-- d_ef: integer (nullable = false)
//  |-- arr_struct: array (nullable = true)
//  |    |-- element: struct (containsNull = true)
//  |    |    |-- a_bc: integer (nullable = false)
//  |    |    |-- d_ef: integer (nullable = false)
//  |-- arr_int: array (nullable = true)
//  |    |-- element: integer (containsNull = false)