scala - 返回动态数据类型的 Apache Spark UDF

Question

我有 UDF处理 JSON 并返回每行的动态数据结果。就我而言，我需要它来验证数据并返回经过验证的数据。

架构对于每一行都是灵活的。这意味着我无法创建 case class对于每种情况(我的一些数据可以嵌套)。

我试图从我的 UDF 函数返回元组，但我也没有运气(因为我需要从列表转换为元组)，我没有找到一个优雅的解决方案。

我返回的数据类型是 String , Integer , Double , DateTime ，以不同的顺序。

我尝试使用 map在 DataFrame 上，但我的架构有问题。

import spark.implicits._

def processData(row_type: String) = {
  /*
  completely random output here. Tuple/List/Array of 
  elements with a type Integer, String, Double, DateType.
  */

  // pseudo-code starts here

  if row_type == A
     (1, "second", 3)
  else
     (1, "second", 3, 4)
}

val processDataUDF = udf((row_type: String) => processData(row_type))

val df = Seq((0, 1), (1, 2)).toDF("a", "b")
val df2 = df.select(processDataUDF($"a"))
df2.show(5)
df2.printSchema()

结果

+------------+
|      UDF(a)|
+------------+
|[1,second,3]|
|[1,second,3]|
+------------+

我应该如何解决这个问题？根据 row_type 我们有不同的处理结果.所有 row_type是动态设置的。我可以很棒 Schema对于每个 row_type ，但我无法使用不同的模式生成相同的 UDF 返回结果。

正在使用 map是这里唯一的方法吗？

Answer 1

Spark Dataset是一个列数据结构，这里真的没有灵活的模式。 Schema 必须是同质的(所有行必须具有相同的一般结构)并且预先知道(如果您使用 UDF，它必须返回定义良好的 SQL 类型)。

您可以通过以下方式获得一些灵 active :

定义表示所有可能字段的超集的模式并将各个列标记为 nullable .这只有在没有类型冲突的情况下才有可能(如果 Row 包含字段 foo 它总是使用相同的 SQL 类型表示)。

使用集合类型(MapType、ArrayType)来表示可变大小的字段。所有值和/或键必须是同一类型。

将原始数据 reshape 到可以用固定模式实际表示的程度。 Spark 包括，作为其依赖项， json4s ，它为 merging, diffing 提供了一组工具和 querying JSON 数据。如果需要，它可用于应用相对复杂的转换。

如果这不切实际，我建议保持 JSON 字段“原样”并仅按需解析以提取特定值。您可以使用 get_json_object和显式类型转换。这允许测试不同的场景:

coalesce(Seq("$.bar", "$.foo.bar", "$.foobar.foo.bar")
  .map(get_json_object($"json_col", _)): _*).cast(DoubleType)

不假设单个文档结构。

您可以使用二进制 Encoders 获得更多的灵 active ( Encoders.kryo , Encoders.java ) 或 RDD API，可用于存储联合类型(甚至 Any )，但如果您真的期望完全随机输出，则表明存在严重的设计或数据建模问题。即使您可以存储解析的数据，也很难使用它。