scala - 如何更改 StructType 的 StructField 中列的数据类型？

Question

我正在尝试更改从 RDBMS 数据库读取的数据框中存在的列的数据类型。为此，我通过以下方式获得了数据框的架构:

val dataSchema = dataDF.schema

为了查看数据框的架构，我使用了以下语句:

println(dataSchema.schema)

Output: StructType(StructField(je_header_id,LongType,true), StructField(je_line_num,LongType,true), StructField(last_update_date,TimestampType,true), StructField(last_updated_by,DecimalType(15,0),true), StructField(creation_date,TimestampType,true), StructField(created_by,DecimalType(15,0),true), StructField(created_by_name,StringType,true), StructField(entered_dr,DecimalType(38,30),true), StructField(entered_cr,DecimalType(38,30),true))

我的要求是从上述架构中找到 DecimalType 并将其更改为 DoubleType。我可以使用以下方法获取列名和数据类型:dataSchema.dtype 但它以 ((columnName1, column datatype),(columnName2, column datatype)....(columnNameN, column datatype) 的格式给我数据类型))

我试图找到一种方法来解析 StructType 并徒劳地更改 dataSchema 中的模式。

任何人都可以让我知道是否有解析 StructType 的方法，以便我可以将数据类型更改为我的要求并获得以下格式

StructType(StructField(je_header_id,LongType,true), StructField(je_line_num,LongType,true), StructField(last_update_date,TimestampType,true), StructField(last_updated_by,DoubleType,true), StructField(creation_date,TimestampType,true), StructField(created_by,DoubleType,true), StructField(created_by_name,StringType,true), StructField(entered_dr,DoubleType,true), StructField(entered_cr,DoubleType,true))

Answer 1

要修改特定于给定数据类型的 DataFrame Schema，您可以对 StructField 进行模式匹配的dataType，如下图:

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

val df = Seq(
  (1L, BigDecimal(12.34), "a", BigDecimal(10.001)),
  (2L, BigDecimal(56.78), "b", BigDecimal(20.002))
).toDF("c1", "c2", "c3", "c4")

val newSchema = df.schema.fields.map{
  case StructField(name, _: DecimalType, nullable, _)
    => StructField(name, DoubleType, nullable)
  case field => field
}
// newSchema: Array[org.apache.spark.sql.types.StructField] = Array(
//   StructField(c1,LongType,false), StructField(c2,DoubleType,true),
//   StructField(c3,StringType,true), StructField(c4,DoubleType,true)
// )

但是，假设您的最终目标是通过更改列类型来转换数据集，那么只遍历目标数据类型的列以迭代地 cast 它们会更容易，如下所示:

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

val df2 = df.dtypes.
  collect{ case (dn, dt) if dt.startsWith("DecimalType") => dn }.
  foldLeft(df)((accDF, c) => accDF.withColumn(c, col(c).cast("Double")))

df2.printSchema
// root
//  |-- c1: long (nullable = false)
//  |-- c2: double (nullable = true)
//  |-- c3: string (nullable = true)
//  |-- c4: double (nullable = true)

[更新]

根据评论的附加要求，如果您只想更改具有正比例的 DecimalType 的架构，只需在方法 guard 中应用 Regex 模式匹配作为条件 收集:

val pattern = """DecimalType\(\d+,(\d+)\)""".r

val df2 = df.dtypes.
  collect{ case (dn, dt) if pattern.findFirstMatchIn(dt).map(_.group(1)).getOrElse("0") != "0" => dn }.
  foldLeft(df)((accDF, c) => accDF.withColumn(c, col(c).cast("Double")))