Querying Spark SQL DataFrame con tipi complessi

Come posso interrogare un RDD con tipi complessi come mappe / array? per esempio, quando stavo scrivendo questo codice di test:

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))) 

Anche se la syntax sarebbe qualcosa di simile:

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

o

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

ma capisco

Imansible accedere al campo nidificato in tipo MapType (StringType, StringType, true)

e

org.apache.spark.sql.catalyst.errors.package $ TreeNodeException: attributi non risolti

rispettivamente.

Dipende da un tipo di colonna. Iniziamo con alcuni dati fittizi:

 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)), 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 
 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) 
  • colonne dell’array:

    • Metodo Column.getItem

       df.select($"an_array".getItem(1)).show // +-----------+ // |an_array[1]| // +-----------+ // | 2| // | 5| // +-----------+ 
    • Sintassi delle parentesi dell’hive:

       sqlContext.sql("SELECT an_array[1] FROM df").show // +---+ // |_c0| // +---+ // | 2| // | 5| // +---+ 
    • un 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| // +---------------+ 
  • colonne della mappa

    • utilizzando il metodo Column.getField :

       df.select($"a_map".getField("foo")).show // +----------+ // |a_map[foo]| // +----------+ // | bar| // | null| // +----------+ 
    • usando la syntax delle parentesi dell’hive:

       sqlContext.sql("SELECT a_map['foz'] FROM df").show // +----+ // | _c0| // +----+ // |null| // | baz| // +----+ 
    • utilizzando un percorso completo con syntax del punto:

       df.select($"a_map.foo").show // +----+ // | foo| // +----+ // | bar| // |null| // +----+ 
    • usando un 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| // +--------------+ 
  • struct columns using full path with dot syntax:

    • con DataFrame API

       df.select($"a_struct.x").show // +---+ // | x| // +---+ // | 1| // | 2| // +---+ 
    • con SQL raw

       sqlContext.sql("SELECT a_struct.x FROM df").show // +---+ // | x| // +---+ // | 1| // | 2| // +---+ 
  • è ansible accedere ai campi all’interno della matrice di structs utilizzando la syntax dei punti, i nomi e i metodi di Column standard:

     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| // +------------------------------+ 
  • è ansible accedere ai campi definiti dall’utente (UDT) utilizzando le UDF. Vedi gli attributi di riferimento di SparkSQL di UDT per i dettagli.

Note :

  • a seconda della versione di Spark alcuni di questi metodi possono essere disponibili solo con HiveContext . Le UDF dovrebbero funzionare indipendentemente dalla versione con SQLContext standard e HiveContext .
  • i valori nidificati in generale sono cittadini di seconda class. Non tutte le operazioni tipiche sono supportate su campi nidificati. A seconda del contesto, potrebbe essere meglio appiattire lo schema e / o esplodere le raccolte

     df.select(explode($"an_array_of_structs")).show // +--------------------+ // | col| // +--------------------+ // |[foo,1,WrappedArr...| // |[bar,2,WrappedArr...| // |[foz,3,WrappedArr...| // |[baz,4,WrappedArr...| // +--------------------+ 
  • La syntax del punto può essere combinata con il carattere jolly ( * ) per selezionare (possibilmente più campi) senza specificare esplicitamente i nomi:

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

Una volta convertito in DF, puoi semplicemente recuperare i dati come

  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") 

ecco cosa ho fatto e ha funzionato

 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 

risultati

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