使用定义的 StructType 转换 Spark 数据帧的值

Posted 2023-04-17

【中文标题】使用定义的 StructType 转换 Spark 数据帧的值

【英文标题】：Cast values of a Spark dataframe using a defined StructType

【发布时间】：2018-07-27 16:07:14

【问题描述】：

有没有办法使用 StructType 转换数据帧的所有值？

让我用一个例子来解释我的问题：

假设我们在读取文件后获得了一个数据帧（我提供了一个生成此数据帧的代码，但在我的实际项目中，我是在从文件读取后获得此数据帧）：

    import org.apache.spark.sql.Row, SparkSession
    import org.apache.spark.sql.types._
    import org.apache.spark.sql.functions._
    import spark.implicits._
    val rows1 = Seq(
      Row("1", Row("a", "b"), "8.00", Row("1","2")),
      Row("2", Row("c", "d"), "9.00", Row("3","4"))
    )

    val rows1Rdd = spark.sparkContext.parallelize(rows1, 4)

    val schema1 = StructType(
      Seq(
        StructField("id", StringType, true),
        StructField("s1", StructType(
          Seq(
            StructField("x", StringType, true),
            StructField("y", StringType, true)
          )
        ), true),
        StructField("d", StringType, true),
        StructField("s2", StructType(
          Seq(
            StructField("u", StringType, true),
            StructField("v", StringType, true)
          )
        ), true)
      )
    )

    val df1 = spark.createDataFrame(rows1Rdd, schema1)

    println("Schema with nested struct")
    df1.printSchema()

    root
    |-- id: string (nullable = true)
    |-- s1: struct (nullable = true)
    |    |-- x: string (nullable = true)
    |    |-- y: string (nullable = true)
    |-- d: string (nullable = true)
    |-- s2: struct (nullable = true)
    |    |-- u: string (nullable = true)
    |    |-- v: string (nullable = true)

现在假设我的客户向我提供了他想要的数据架构（相当于读取数据帧的架构，但具有不同的数据类型（包含 StringTypes、IntegerTypes ...））：

    val wantedSchema = StructType(
      Seq(
        StructField("id", IntegerType, true),
        StructField("s1", StructType(
          Seq(
            StructField("x", StringType, true),
            StructField("y", StringType, true)
          )
        ), true),
        StructField("d", DoubleType, true),
        StructField("s2", StructType(
          Seq(
            StructField("u", IntegerType, true),
            StructField("v", IntegerType, true)
          )
        ), true)
      )
    )

使用提供的 StructType 转换数据框值的最佳方法是什么？

如果有一种方法可以应用于数据帧，那就太好了，它通过自己转换所有值来应用新的 StructTypes。

PS：这是一个小数据框，用作示例，在我的项目中，数据框包含更多行。 如果它是一个包含几列的小型 Dataframe，我可以轻松完成转换，但就我而言，我正在寻找一种智能解决方案，通过应用 StructType 来转换所有值，而无需手动转换每个列/值代码。

如果您能提供任何帮助，我将不胜感激，非常感谢！

【参考方案1】：

经过大量研究，这里有一个通用的解决方案，可以按照模式转换数据框：

val castedDf = df1.selectExpr(wantedSchema.map(
  field => s"CAST ( $field.name As $field.dataType.sql) $field.name"
): _*)

这是转换后的数据框的架构：

castedDf.printSchema
root
|-- id: integer (nullable = true)
|-- s1: struct (nullable = true)
|    |-- x: string (nullable = true)
|    |-- y: string (nullable = true)
|-- d: double (nullable = true)
|-- s2: struct (nullable = true)
|    |-- u: integer (nullable = true)
|    |-- v: integer (nullable = true)

我希望它能对某人有所帮助，我花了 5 天时间寻找这个简单/通用的解决方案。

【讨论】：

您可能想要添加处理STRUCT 类型的逻辑 - 正如它所写的那样，此代码甚至无法完全处理您的原始示例。除此之外，您走在正确的轨道上 - 尽管您仍然需要在某处指定您的映射，因此它不完全是“通用”解决方案

其实它可以处理STRUCT类型、数组、结构数组等...我邀请你测试一下。我对此进行了多次测试。我有一个关于 apache spark 的拉取请求，以便将其添加为数据框函数（我希望它会被合并）。

【参考方案2】：

没有自动执行转换的方法。您可以在 Spark SQL 中表达转换逻辑，一次性转换所有内容 - 但是，如果您有很多字段，生成的 SQL 可能会变得非常大。但至少你可以将所有的转变都集中在一个地方。

例子：

   df1.selectExpr("CAST (id AS INTEGER) as id",
    "STRUCT (s1.x, s1.y) AS s1",
    "CAST (d AS DECIMAL) as d",
    "STRUCT (CAST (s2.u AS INTEGER), CAST (s2.v AS INTEGER)) as s2").show()

需要注意的一点是，只要转换失败（例如，当d 不是数字时），您就会得到一个NULL。一种选择是在转换之前运行一些验证，然后过滤掉 df1 记录以仅转换有效的记录。

【讨论】：

感谢您的回答.. 我正在寻找一个解决方案，我不必为每列编写铸件。可惜它不存在

我找到了以通用方式执行转换的解决方案，我会在答案中发布。