spark 不支持的操作异常：不支持 org.apache.spark.sql.types.DataType 类型的架构

Question

Spark 流式传输：

我收到一个包含两列的数据框。第一列是string 类型，包含一个json 字符串，第二列由schema 组成，每个value（第一列）。

Batch: 0
-------------------------------------------
+--------------------+--------------------+
|               value|              schema|
+--------------------+--------------------+
|{"event_time_abc...|`event_time_abc...|
+--------------------+--------------------+

该表存储在val input（非可变变量）中。我正在使用DataType.fromDDL 函数通过以下方式将string 类型转换为json DataFrame：

val out=  input.select(from_json(col("value").cast("string"),ddl(col("schema"))))

ddl 是一个预定义函数，DataType.from_ddl(_:String):DataType 在 spark(scala) 中，但我已经注册了它，以便我可以在整个列上使用它，而不是仅在字符串上使用它。我是通过以下方式完成的：

val ddl:UserDefinedFunction = udf(DataType.fromDDL(_:String):DataType)

这是input 表的列、值和架构的最终转换。

val out =  input.select(from_json(col("value").cast("string"),ddl(col("schema"))))

但是，我在这一行的注册中遇到了异常：

val ddl:UserDefinedFunction = udf(DataType.fromDDL(_:String):DataType)

错误是：

java.lang.UnsupportedOperationException: Schema for type org.apache.spark.sql.types.DataType is not supported
 

如果我使用：

val out =  input.select(from_json(col("value").cast("string"),DataType.fromDDL("`" + "event_time_human2"+"`" +" STRING")).alias("value"))

然后它可以工作，但正如你所见，我只在函数DataType.fromDDL(_:String):DataType 中使用string（手动输入来自schema 列）。

那么如何在不注册的情况下将此功能应用于整个列，或者有没有其他方法可以注册该功能？

编辑：from_json 函数的第一个参数需要一个列，而第二个参数需要一个 schema 而不是一个列。因此，我想需要一种手动方法来解析每个 value 字段和每个 schema 字段。经过一番调查，我发现 DataFrames 不支持 DataType。

---

自从在这个问题上设置了赏金。我想提供有关数据和架构的其他信息。模式在 DDL（字符串类型）中定义，可以使用 from_DDL 函数解析。该值是简单的json 字符串，将使用我们使用from_DDL 函数派生的架构进行解析。

基本思想是每个值都有自己的模式，需要用相应的模式进行解析。应该在存储结果的位置创建一个新列。

数据： 这是数据的一个示例：

值 = {"event_time_human2":"09:45:00 +0200 09/27/2021"}

schema = "`event_time_human2` STRING"

不需要转换为正确的时间格式。只要一个字符串就可以了。

它位于streaming context。因此，并非所有方法都有效。

Answer 1

在运行时之前应用和验证模式，即在执行器上执行 Spark 代码之前。解析的模式必须是执行计划的一部分，因此模式解析直到现在还不能按照您的预期动态执行。这就是您看到异常的原因： java.lang.UnsupportedOperationException: Schema for type org.apache.spark.sql.types.DataType is not supported 仅适用于 UDF。因此，这意味着 DataType.fromDDL 应仅在驱动程序代码中使用，而不应在运行时/执行程序代码中使用，后者是 UDF 函数中的代码。在 UDF 函数内，Spark 已经应用您在驱动程序端指定的模式执行了导入数据的转换。这就是您不能在 UDF 中直接使用 DataType.fromDDL 的原因，因为它本质上是无用的。以上所有意味着在 UDF 函数内部，我们只能使用原始的 Scala/Java 类型和 Spark API 提供的一些包装器，即 WrappedArray。

另一种方法是收集驱动程序上的所有模式。然后为每个模式创建一个包含对（模式、数据框）的映射。

请记住，向驱动程序收集数据是一项昂贵的操作，只有当您拥有合理数量的唯一模式（即最多数千个）时才有意义。此外，将这些模式应用于每个数据集需要在驱动程序中按顺序完成，这也非常昂贵，因此重要的是要认识到建议的解决方案只有在唯一模式数量有限的情况下才能有效工作。

到目前为止，您的代码可能如下所示：

import org.apache.spark.sql.functions.from_json
import org.apache.spark.sql.types.StructType

import spark.implicits._

val df = Seq(
  ("""{"event_time_human2":"09:45:00 +0200 09/27/2021", "name":"Pinelopi"}""", "`event_time_human2` STRING, name STRING"),
  ("""{"first_name":"Pin", "last_name":"Halen"}""", "first_name STRING, last_name STRING"),
  ("""{"code":993, "full_name":"Sofia Loren"}""", "code INT, full_name STRING")
).toDF("value", "schema")

val schemaDf = df.select("schema").distinct()

val dfBySchema = schemaDf.collect().map{ row =>
  val schemaValue = row.getString(0)
  val ddl = StructType.fromDDL(schemaValue)
  val filteredDf = df.where($"schema" === schemaValue)
                     .withColumn("value", from_json($"value", ddl))
  
  (schemaValue, filteredDf)
}.toMap

// Map(
//   `event_time_human2` STRING, name STRING -> [value: struct<event_time_human2: string, name: string>, schema: string], 
//   first_name STRING, last_name STRING -> [value: struct<first_name: string, last_name: string>, schema: string], 
//   code INT, full_name STRING -> [value: struct<code: int, full_name: string>, schema: string]
// )

说明：首先我们用schemaDf.collect() 收集每个独特的模式。然后我们遍历模式并根据当前模式过滤初始 df。我们还使用from_json 将当前字符串值列转换为特定架构。

注意，我们不能有一个具有不同数据类型的公共列，这就是我们为每个架构创建不同的 df 而不是最终的 df 的原因。