Spark SQLのDataFrameを複雑な型でクエリする

2023-08-13 21:31:53

質問

マップや配列のような複雑な型を持つ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: 未解決の属性

をそれぞれ追加しました。

どのように解決するのですか？

カラムの種類に依存します。まずはダミーデータから見てみましょう。

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

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 (SQL2.4+、Scala3.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 ) のカラムをドット構文でフルパスで指定します。

データフレーム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|
// +------------------------------+

ユーザー定義型（UDT）のフィールドは、UDFを使用してアクセスすることができます。参照 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|
// +---+
```
JSONカラムの問い合わせは get_json_object と from_json 関数を使用することができます。参照 Spark DataFramesを使用してJSONデータカラムをクエリする方法は？を参照してください。

Spark SQLのDataFrameを複雑な型でクエリする

質問

どのように解決するのですか？

関連

[解決済み] SQL ServerでJOINを使用してUPDATE文を実行するにはどうすればよいですか？

[解決済み] PostgreSQLからのPL/pgSQL出力をCSVファイルに保存する

[解決済み] R dataframeでNAの値をゼロに置き換えるには？

[解決済み] SQL SELECT WHERE フィールドに単語が含まれる場合

[解決済み] ある列の最大値を持つ行を取得する

[解決済み] SQLite - UPSERT not INSERT or REPLACE

[解決済み] 項目xにアクセスできるように文字列を分割するにはどうすればよいですか？

[解決済み】SQL Serverで既存のテーブルにデフォルト値を持つカラムを追加する

[解決済み] SQL ServerでINNER JOINを使用して削除するにはどうすればよいですか？

[解決済み】SparkのDataFrame、Dataset、RDDの違いについて

最新

nginxです。[emerg] 0.0.0.0:80 への bind() に失敗しました (98: アドレスは既に使用中です)

htmlページでギリシャ文字を使うには

ピュアhtml+cssでの要素読み込み効果

純粋なhtml + cssで五輪を実現するサンプルコード

ナビゲーションバー・ドロップダウンメニューのHTML+CSSサンプルコード

タイピング効果を実現するピュアhtml+css

htmlの選択ボックスのプレースホルダー作成に関する質問

html css3 伸縮しない画像表示効果

トップナビゲーションバーメニュー作成用HTML+CSS

html+css 実装サイバーパンク風ボタン

おすすめ

[解決済み] SQL Serverで実行中の合計を計算する

[解決済み] アドホッククエリとは何ですか？

[解決済み] SQL ServerでSELECTからUPDATEする方法とは？

[解決済み] 各GROUP BYグループの最初の行を選択しますか？

[解決済み] 複数の列でgroup byを使用する

[解決済み] SQLの複数列の順序付け

[解決済み] DISTINCTでCOUNT(*)を選択する

[解決済み] 既存のカラムにIDを追加する

[解決済み] T-SQL文の接頭辞Nの意味と使うべきタイミングは？

[解決済み] Selectステートメントで特定のフィールドの重複を検索する