查询具有复杂类型的Spark SQL DataFrame

小编典典

查询具有复杂类型的Spark SQL DataFrame

sql

如何查询具有复杂类型（如地图/数组）的RDD？例如，当我编写此测试代码时：

case class Test(name: String, map: Map[String, String])
val map = Map("hello" -> "world", "hey" -> "there")
val map2 = Map("hello" -> "people", "hey" -> "you")
val rdd = sc.parallelize(Array(Test("first", map), Test("second", map2)))

我认为语法应该是这样的：

sqlContext.sql("SELECT * FROM rdd WHERE map.hello = world")

或者

sqlContext.sql("SELECT * FROM rdd WHERE map[hello] = world")

但是我明白了

无法访问类型为MapType（StringType，StringType，true）的嵌套字段

和

org.apache.spark.sql.catalyst.errors.package $ TreeNodeException：无法解析的属性

分别。

阅读 261

2021-05-05

共1个答案

小编典典

它取决于列的类型。让我们从一些虚拟数据开始：

import org.apache.spark.sql.functions.{udf, lit}
import scala.util.Try

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, 2.0)),
           ArrayElement("bar", 2, Array(3.0, 4.0, 5.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



df.registerTempTable("df")
df.printSchema

// root
// |-- an_array: array (nullable = true)
// |    |-- element: integer (containsNull = false)
// |-- a_map: map (nullable = true)
// |    |-- key: string
// |    |-- value: string (valueContainsNull = true)
// |-- a_struct: struct (nullable = true)
// |    |-- x: integer (nullable = false)
// |-- an_array_of_structs: array (nullable = true)
// |    |-- element: struct (containsNull = true)
// |    |    |-- foo: string (nullable = true)
// |    |    |-- bar: integer (nullable = false)
// |    |    |-- vals: array (nullable = true)
// |    |    |    |-- element: double (containsNull = false)

数组（ArrayType）列：
- Column.getItem 方法
```
    df.select($"an_array".getItem(1)).show
```
  // +-----------+
  // |an_array[1]|
  // +-----------+
  // | 2|
  // | 5|
  // +-----------+
- 蜂巢括号语法：
```
    sqlContext.sql("SELECT an_array[1] FROM df").show
```
  // +—+
  // |_c0|
  // +—+
  // | 2|
  // | 5|
  // +—+
- 一个UDF
```
    val get_ith = udf((xs: Seq[Int], i: Int) => Try(xs(i)).toOption)
```
  df.select(get_ith($”an_array”, lit(1))).show
  
  // +---------------+
  // |UDF(an_array,1)|
  // +---------------+
  // | 2|
  // | 5|
  // +---------------+
- 除了上面列出的方法外，Spark还支持越来越多的对复杂类型进行操作的内置函数。值得注意的示例包括更高阶的函数，例如transform（SQL 2.4 +，Scala 3.0 +，PySpark / SparkR 3.1+）：
```
    df.selectExpr("transform(an_array, x -> x + 1) an_array_inc").show
```
  // +------------+
  // |an_array_inc|
  // +------------+
  // | [2, 3, 4]|
  // | [5, 6, 7]|
  // +------------+
  
  import org.apache.spark.sql.functions.transform
  
  df.select(transform($”an_array”, x => x + 1) as “an_array_inc”).show
  // +------------+
  // |an_array_inc|
  // +------------+
  // | [2, 3, 4]|
  // | [5, 6, 7]|
  // +------------+
- filter （SQL 2.4以上版本，Scala 3.0以上版本，Python / SparkR 3.1以上版本）
```
    df.selectExpr("filter(an_array, x -> x % 2 == 0) an_array_even").show
```
  // +-------------+
  // |an_array_even|
  // +-------------+
  // | [2]|
  // | [4, 6]|
  // +-------------+
  
  import org.apache.spark.sql.functions.filter
  
  df.select(filter($”an_array”, x => x % 2 === 0) as “an_array_even”).show
  // +-------------+
  // |an_array_even|
  // +-------------+
  // | [2]|
  // | [4, 6]|
  // +-------------+
- aggregate （SQL 2.4以上版本，Scala 3.0以上版本，PySpark / SparkR 3.1以上版本）：
```
    df.selectExpr("aggregate(an_array, 0, (acc, x) -> acc + x, acc -> acc) an_array_sum").show
```
  // +------------+
  // |an_array_sum|
  // +------------+
  // | 6|
  // | 15|
  // +------------+
  
  import org.apache.spark.sql.functions.aggregate
  
  df.select(aggregate($”an_array”, lit(0), (x, y) => x + y) as “an_array_sum”).show
  // +------------+
  // |an_array_sum|
  // +------------+
  // | 6|
  // | 15|
  // +------------+
- 数组处理函数（array_*），如array_distinct（2.4+）：
```
    import org.apache.spark.sql.functions.array_distinct
```
  df.select(array_distinct($”an_array_of_structs.vals”(0))).show
  // +-------------------------------------------+
  // |array_distinct(an_array_of_structs.vals[0])|
  // +-------------------------------------------+
  // | [1.0, 2.0]|
  // | [5.0, 6.0]|
  // +-------------------------------------------+
- array_max（array_min，2.4+）：
```
    import org.apache.spark.sql.functions.array_max
```
  df.select(array_max($”an_array”)).show
  // +-------------------+
  // |array_max(an_array)|
  // +-------------------+
  // | 3|
  // | 6|
  // +-------------------+
- flatten （2.4+）
```
    import org.apache.spark.sql.functions.flatten
```
  df.select(flatten($”an_array_of_structs.vals”)).show
  // +---------------------------------+
  // |flatten(an_array_of_structs.vals)|
  // +---------------------------------+
  // | [1.0, 2.0, 2.0, 3…|
  // | [5.0, 6.0, 7.0, 8.0]|
  // +---------------------------------+
- arrays_zip （2.4+）：
```
    import org.apache.spark.sql.functions.arrays_zip
```
  df.select(arrays_zip($”an_array_of_structs.vals”(0), $”an_array_of_structs.vals”(1))).show(false)
  // +--------------------------------------------------------------------+
  // |arrays_zip(an_array_of_structs.vals[0], an_array_of_structs.vals[1])|
  // +--------------------------------------------------------------------+
  // |[[1.0, 3.0], [2.0, 4.0], [2.0, 5.0]] |
  // |[[5.0, 7.0], [6.0, 8.0]] |
  // +--------------------------------------------------------------------+
- array_union （2.4+）：
```
    import org.apache.spark.sql.functions.array_union
```
  df.select(array_union($”an_array_of_structs.vals”(0), $”an_array_of_structs.vals”(1))).show
  // +---------------------------------------------------------------------+
  // |array_union(an_array_of_structs.vals[0], an_array_of_structs.vals[1])|
  // +---------------------------------------------------------------------+
  // | [1.0, 2.0, 3.0, 4…|
  // | [5.0, 6.0, 7.0, 8.0]|
  // +---------------------------------------------------------------------+
- slice （2.4+）：
```
    import org.apache.spark.sql.functions.slice
```
  df.select(slice($”an_array”, 2, 2)).show
  // +---------------------+
  // |slice(an_array, 2, 2)|
  // +---------------------+
  // | [2, 3]|
  // | [5, 6]|
  // +---------------------+
地图（MapType）栏
- 使用Column.getField方法：
```
    df.select($"a_map".getField("foo")).show
```
  // +----------+
  // |a_map[foo]|
  // +----------+
  // | bar|
  // | null|
  // +----------+
- 使用Hive方括号语法：
```
    sqlContext.sql("SELECT a_map['foz'] FROM df").show
```
  // +----+
  // | _c0|
  // +----+
  // |null|
  // | baz|
  // +----+
- 使用带点语法的完整路径：
```
    df.select($"a_map.foo").show
```
  // +----+
  // | foo|
  // +----+
  // | bar|
  // |null|
  // +----+
- 使用UDF
```
    val get_field = udf((kvs: Map[String, String], k: String) => kvs.get(k))
```
  df.select(get_field($”a_map”, lit(“foo”))).show
  
  // +--------------+
  // |UDF(a_map,foo)|
  // +--------------+
  // | bar|
  // | null|
  // +--------------+
- 越来越多的map_*功能，例如map_keys（2.3+）
```
    import org.apache.spark.sql.functions.map_keys
```
  df.select(map_keys($”a_map”)).show
  // +---------------+
  // |map_keys(a_map)|
  // +---------------+
  // | [foo]|
  // | [foz]|
  // +---------------+
- 或map_values（2.3+）
```
    import org.apache.spark.sql.functions.map_values
```
  df.select(map_values($”a_map”)).show
  // +-----------------+
  // |map_values(a_map)|
  // +-----------------+
  // | [bar]|
  // | [baz]|
  // +-----------------+

请检查SPARK-23899以获取详细列表。

StructType使用带有点语法的完整路径的struct（）列：
- 使用DataFrame API
```
    df.select($"a_struct.x").show
```
  // +—+
  // | x|
  // +—+
  // | 1|
  // | 2|
  // +—+
- 用原始SQL
```
    sqlContext.sql("SELECT a_struct.x FROM df").show
```
  // +—+
  // | x|
  // +—+
  // | 1|
  // | 2|
  // +—+
数组中的字段structs可以使用点语法，名称和标准Column方法进行访问：
```
df.select($"an_array_of_structs.foo").show
```
// +----------+
// | foo|
// +----------+
// |[foo, bar]|
// |[foz, baz]|
// +----------+

sqlContext.sql(“SELECT an_array_of_structs[0].foo FROM df”).show

// +—+
// |_c0|
// +—+
// |foo|
// |foz|
// +—+

df.select($”an_array_of_structs.vals”.getItem(1).getItem(1)).show

// +------------------------------+
// |an_array_of_structs.vals[1][1]|
// +------------------------------+
// | 4.0|
// | 8.0|
// +------------------------------+
可以使用UDF访问用户定义的类型（UDT）字段。有关详细信息，请参见UDT的Spark SQL引用属性。

注意事项 ：

根据Spark版本的不同，其中某些方法仅可用于HiveContext。UDF的工作应该独立的版本，采用标准的SQLContext和HiveContext。
一般来说，嵌套的价值观是二等公民。嵌套字段上不支持所有典型操作。根据上下文，最好展平架构和/或爆炸集合
```
df.select(explode($"an_array_of_structs")).show
```
// +--------------------+
// | col|
// +--------------------+
// |[foo,1,WrappedArr…|
// |[bar,2,WrappedArr…|
// |[foz,3,WrappedArr…|
// |[baz,4,WrappedArr…|
// +--------------------+
点语法可以与通配符（*）结合使用以选择（可能是多个）字段，而无需明确指定名称：
```
df.select($"a_struct.*").show
```
// +—+
// | x|
// +—+
// | 1|
// | 2|
// +—+
可以使用get_json_object和from_json函数查询JSON列。

2021-05-05