当前位置：theitroad>使用复杂类型查询 Spark SQL DataFrame

使用复杂类型查询 Spark SQL DataFrame

声明：本页面是StackOverFlow热门问题的中英对照翻译，遵循CC BY-SA 4.0协议，如果您需要使用它，必须同样遵循CC BY-SA许可，注明原文地址和作者信息，同时你必须将它归于原作者（不是我）：StackOverFlow 原文地址: http://stackoverflow.com/questions/28332494/
Warning: these are provided under cc-by-sa 4.0 license. You are free to use/share it, But you must attribute it to the original authors (not me): StackOverFlow

提示：将鼠标放在中文语句上可以显示对应的英文。显示中英文
时间：2020-09-01 03:14:23 　来源:igfitidea点击:

Querying Spark SQL DataFrame with complex types

sqlscalaapache-sparkdataframeapache-spark-sql

提问by dvir

How Can I query an RDD with complex types such as maps/arrays? for example, when I was writing this test code:

如何查询具有复杂类型（例如映射/数组）的 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)))

I thought the syntax would be something like:

我认为语法会是这样的：

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

or

或者

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

but I get

但我明白了

Can't access nested field in type MapType(StringType,StringType,true)

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

and

和

org.apache.spark.sql.catalyst.errors.package$TreeNodeException: Unresolved attributes

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

respectively.

分别。

回答by zero323

It depends on a type of the column. Lets start with some dummy data:

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

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)

  • array (ArrayType) columns:

    • Column.getItemmethod

      df.select($"an_array".getItem(1)).show

// +-----------+
// |an_array[1]|
// +-----------+
// |          2|
// |          5|
// +-----------+

    • Hive brackets syntax:

      sqlContext.sql("SELECT an_array[1] FROM df").show

// +---+
// |_c0|
// +---+
// |  2|
// |  5|
// +---+

    • an 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|
// +---------------+

    • Additionally to the methods listed above Spark supports a growing list of built-in functions operating on complex types. Notable examples include higher order functions like 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 processing functions (array_*) like 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) columns

    • using Column.getFieldmethod:

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

// +----------+
// |a_map[foo]|
// +----------+
// |       bar|
// |      null|
// +----------+

    • using Hive brackets syntax:

      sqlContext.sql("SELECT a_map['foz'] FROM df").show

// +----+
// | _c0|
// +----+
// |null|
// | baz|
// +----+

    • using a full path with dot syntax:

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

// +----+
// | foo|
// +----+
// | bar|
// |null|
// +----+

    • using an 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|
// +--------------+

    • Growing number of map_*functions like 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]|
// +---------------+

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

    Please check SPARK-23899for a detailed list.

  • struct (StructType) columns using full path with dot syntax:

    • with DataFrame API

      df.select($"a_struct.x").show

// +---+
// |  x|
// +---+
// |  1|
// |  2|
// +---+

    • with raw SQL

      sqlContext.sql("SELECT a_struct.x FROM df").show

// +---+
// |  x|
// +---+
// |  1|
// |  2|
// +---+

  • fields inside array of structscan be accessed using dot-syntax, names and standard Columnmethods:

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

  • user defined types (UDTs) fields can be accessed using UDFs. See Spark SQL referencing attributes of UDTfor details.

  • 数组 ( ArrayType) 列：

    • Column.getItem方法

      df.select($"an_array".getItem(1)).show

// +-----------+
// |an_array[1]|
// +-----------+
// |          2|
// |          5|
// +-----------+

    • Hive 括号语法：

      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以获取详细列表。

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

    • 使用数据帧 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) 字段。有关详细信息，请参阅Spark SQL 引用 UDT 的属性。

Notes:

注意事项：

  • depending on a Spark version some of these methods can be available only with HiveContext. UDFs should work independent of version with both standard SQLContextand HiveContext.

  • generally speaking nested values are a second class citizens. Not all typical operations are supported on nested fields. Depending on a context it could be better to flatten the schema and / or explode collections

    df.select(explode($"an_array_of_structs")).show

// +--------------------+
// |                 col|
// +--------------------+
// |[foo,1,WrappedArr...|
// |[bar,2,WrappedArr...|
// |[foz,3,WrappedArr...|
// |[baz,4,WrappedArr...|
// +--------------------+

  • Dot syntax can be combined with wildcard character (*) to select (possibly multiple) fields without specifying names explicitly:

    df.select($"a_struct.*").show
// +---+
// |  x|
// +---+
// |  1|
// |  2|
// +---+

  • JSON columns can be queried using get_json_objectand from_jsonfunctions. See How to query JSON data column using Spark DataFrames?for details.

  • 根据 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 列。请参阅如何使用 Spark DataFrames 查询 JSON 数据列？详情。

回答by sshroff

Once You convert it to DF, u can simply fetch data as

一旦您将其转换为 DF，您就可以简单地将数据作为 

  val rddRow= rdd.map(kv=>{
    val k = kv._1
    val v = kv._2
    Row(k, v)
  })

val myFld1 =  StructField("name", org.apache.spark.sql.types.StringType, true)
val myFld2 =  StructField("map", org.apache.spark.sql.types.MapType(StringType, StringType), true)
val arr = Array( myFld1, myFld2)
val schema = StructType( arr )
val rowrddDF = sqc.createDataFrame(rddRow, schema)
rowrddDF.registerTempTable("rowtbl")  
val rowrddDFFinal = rowrddDF.select(rowrddDF("map.one"))
or
val rowrddDFFinal = rowrddDF.select("map.one")

回答by Sumit Pal

here was what I did and it worked

这就是我所做的并且有效

case class Test(name: String, m: 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)))
val rdddf = rdd.toDF
rdddf.registerTempTable("mytable")
sqlContext.sql("select m.hello from mytable").show

Results

结果

+------+
| hello|
+------+
| world|
|people|
+------+

相关推荐

SQL 通过 SELECT 语句更新多个字段 FROM

SQL 通过 SELECT 语句更新多个字段 FROM

SQL Oracle - 用句号表示千位，用逗号表示小数点

SQL Oracle - 用句号表示千位，用逗号表示小数点

SQL Zend Framework Db 选择连接表帮助

SQL Zend Framework Db 选择连接表帮助

SQL 您如何解决 MS Access 2010 中的“条件表达式中的数据类型不匹配”错误？

SQL 您如何解决 MS Access 2010 中的“条件表达式中的数据类型不匹配”错误？

相关推荐
最近更新
标签