动态查询复杂类型的spark sql数据帧

Question

我试图用复杂类型查询spark sql数据帧，其中函数本身应该能够创建和表达式来为嵌套的复杂数据类型生成列数据帧。说

case class SubRecord(x: Int)
case class ArrayElement(foo: String, bar: Int, vals: Array[Double])
case class Record(
  an_array: Array[Int], a_map: Map[String, String], 
  a_struct: SubRecord, an_array_of_structs: Array[ArrayElement])

val df = sc.parallelize(Seq(
  Record(Array(1, 2, 3), Map("foo" -> "bar"), SubRecord(1),
         Array(
           ArrayElement("foo", 1, Array(1.0, 2.0)),
           ArrayElement("bar", 2, Array(3.0, 4.0)))),
  Record(Array(4, 5, 6), Map("foz" -> "baz"), SubRecord(2),
         Array(ArrayElement("foz", 3, Array(5.0, 6.0)), 
               ArrayElement("baz", 4, Array(7.0, 8.0))))
)).toDF

引用自查询SparkSQL具有复杂类型的DataFrame

用于提取地图类型查询可以是

df.select($"a_map.foo").show

现在如果我有

case class Record(
  an_array: Array[Int], a_map_new: Map[String, Array[ArrayElement]], 
  a_struct: SubRecord, an_array_of_structs: Array[ArrayElement])

代替Map[String，String],如何创建一个udf，在数组的情况下接受名称或索引，并为复杂数据类型中的嵌套元素生成结果。假设现在我想查询a_map_new中包含的< code > val[0]。

Answer 1

在这种情况下，如果您有定义良好的记录类型，我建议使用强类型＜code＞数据集：

val result = df.as[Record].map(_.a_map_new.mapValues(_.headOption))

result.printSchema
// root
//  |-- value: map (nullable = true)
//  |    |-- key: string
//  |    |-- value: struct (valueContainsNull = true)
//  |    |    |-- foo: string (nullable = true)
//  |    |    |-- bar: integer (nullable = false)
//  |    |    |-- vals: array (nullable = true)
//  |    |    |    |-- element: double (containsNull = false)

使用udf的障碍是它的不对称性质：

• 它获取内部类型＜code＞行＜code＞
• 应返回外部类型
• 必须具有静态定义的返回类型

任何返回结构的通用解决方案都应该知道如何将映射到外部类型。我想你可以设计这样的东西（伪代码）

def f(mapper: Row => T) = udf((map: Map[U, Row]) => map.mapValues(mapper(_)))