如何查询具有复杂类型(如映射/数组)的RDD?

case class Test(name: String, map: Map[String, String])
val map = Map("hello" -> "w或ld", "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 = w或ld")

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

但我明白

无法访问类型MapType(StringType、StringType、true)中的嵌套字段

或g.apache.spark.sql.catalyst.err或s.package$TreeNodeException: Unresolved attributes

分别.

推荐答案

这取决于柱的类型.让我们从一些虚拟数据开始:

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|
      // +---+
      
    • 自由民主党

      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]|
      // +---------------------+
      
  • map (MapType)栏

    • 使用Column.getField法:

      df.select($"a_map".getField("foo")).show
      
      // +----------+
      // |a_map[foo]|
      // +----------+
      // |       bar|
      // |      null|
      // +----------+
      
    • using 配置单元括号语法:

      sqlContext.sql("SELECT a_map['foz'] FROM df").show
      
      // +----+
      // | _c0|
      // +----+
      // |null|
      // | baz|
      // +----+
      
    • 使用带点语法的完整路径:

      df.select($"a_map.foo").show
      
      // +----+
      // | foo|
      // +----+
      // | bar|
      // |null|
      // +----+
      
    • using 自由民主党

      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]|
      // +-----------------+
      

    详细 list 请勾选SPARK-23899.

  • 使用带点语法的完整路径的struct(StructType)列:

    • 使用DataFrame API

      df.select($"a_struct.x").show
      
      // +---+
      // |  x|
      // +---+
      // |  1|
      // |  2|
      // +---+
      
    • 使用原始SQL

      sqlContext.sql("SELECT a_struct.x FROM df").show
      
      // +---+
      // |  x|
      // +---+
      // |  1|
      // |  2|
      // +---+
      
  • 可以使用点语法、名称和标准Column方法访问structs数组中的字段:

    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|
    // +------------------------------+
    
  • 用户定义类型(UDT)字段可以使用UDF访问.详情见Spark SQL referencing attributes of UDT.

Notes:

  • 根据Spark版本的不同,这些方法中的一些仅适用于HiveContext.UDF应该独立于标准SQLContextHiveContext的版本.
  • 一般来说,嵌套值是二等公民.并不是所有的典型操作都支持嵌套字段.根据上下文的不同,可以更好地展平模式和/或分解集合

    df.select(explode($"an_array_of_structs")).show
    
    // +--------------------+
    // |                 col|
    // +--------------------+
    // |[foo,1,WrappedArr...|
    // |[bar,2,WrappedArr...|
    // |[foz,3,WrappedArr...|
    // |[baz,4,WrappedArr...|
    // +--------------------+
    
  • 点语法可以与通配符(*)结合使用,以 Select (可能有多个)字段,而无需明确指定名称:

    df.select($"a_struct.*").show
    // +---+
    // |  x|
    // +---+
    // |  1|
    // |  2|
    // +---+
    
  • 可以使用get_json_objectfrom_json函数查询JSON列.详情见How to query JSON data column using Spark DataFrames?.

Sql相关问答推荐

SQL使最终结果显示两个表后的所有数据(匹配和不匹配)?

解析键-值对,根据值 Select ,并使用SQL创建新列

Lag()函数的差异:R与SQL(将R代码转换为SQL)

更新PostgreSQL 15中的JSON值

PostgreSQL使用SQL子查询在时间间隔内 Select 数据

PostgreSQL 9.6嵌套的INSERT/RETURN语句的CTE性能低得令人无法接受

使用generate_series()时,LEFT联接缺少日期/间隔

Oracle 23c ROUND,数据类型为DATE

Oracle PL/SQL:解决DBMS输出大小限制的问题

向表中添加新列取决于表的日期列(unpivot)

SAS proc freq 或 proc sql 获取数据子集和整个数据的频率

我可以在 T-SQL (SQL Server) 的函数内使用 OPTION 子句吗?

Oracle PL/SQL长期运行问题

try 将多行折叠为单个结果

检索具有相同位置的最小和最大store 数量

Postgres 条件求和函数

如何在 SQL Server 中参数化 Select top 'n'

为什么 PostgreSQL 1 年间隔 <> 365 天间隔?

在 SQL 中加入 2 个表并根据匹配创建一个标志列

具有 GROUP BY 和 HAVING 的 BIGQUERY SQL 代码以过滤掉观察结果