Consulta de Spark SQL DataFrame con tipos complejos

¿Cómo puedo consultar un RDD con tipos complejos como mapas / matrices? por ejemplo, cuando escribía este código de prueba:

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

Pensé que la syntax sería algo así como:

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

o

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

pero yo obtengo

No se puede acceder al campo nested en tipo MapType (StringType, StringType, true)

y

org.apache.spark.sql.catalyst.errors.package $ TreeNodeException: atributos no resueltos

respectivamente.

Depende de un tipo de columna. Comencemos con algunos datos ficticios:

 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) 
  • columnas de matriz:

    • Método Column.getItem

       df.select($"an_array".getItem(1)).show // +-----------+ // |an_array[1]| // +-----------+ // | 2| // | 5| // +-----------+ 
    • Sintaxis de paréntesis de paréntesis

       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| // +---------------+ 
  • columnas de mapa

    • utilizando el método Column.getField :

       df.select($"a_map".getField("foo")).show // +----------+ // |a_map[foo]| // +----------+ // | bar| // | null| // +----------+ 
    • utilizando la syntax de los paréntesis de Hive:

       sqlContext.sql("SELECT a_map['foz'] FROM df").show // +----+ // | _c0| // +----+ // |null| // | baz| // +----+ 
    • usando una ruta completa con syntax de puntos:

       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| // +--------------+ 
  • columnas struct usando ruta completa con syntax de punto:

    • con la API DataFrame

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

       sqlContext.sql("SELECT a_struct.x FROM df").show // +---+ // | x| // +---+ // | 1| // | 2| // +---+ 
  • Se puede acceder a los campos dentro de la matriz de structs usando syntax de punto, nombres y métodos de Column estándar:

     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| // +------------------------------+ 
  • Se puede acceder a los campos de tipos definidos por el usuario (UDT) usando UDF. Vea los atributos de referencia de SparkSQL de UDT para más detalles.

Notas :

  • dependiendo de la versión de Spark, algunos de estos métodos solo pueden estar disponibles con HiveContext . Las UDF deben funcionar independientemente de la versión con SQLContext estándar y HiveContext .
  • en general, los valores nesteds son ciudadanos de segunda clase. No todas las operaciones típicas son compatibles con los campos nesteds. Dependiendo de un contexto, podría ser mejor aplanar el esquema y / o explotar las colecciones

     df.select(explode($"an_array_of_structs")).show // +--------------------+ // | col| // +--------------------+ // |[foo,1,WrappedArr...| // |[bar,2,WrappedArr...| // |[foz,3,WrappedArr...| // |[baz,4,WrappedArr...| // +--------------------+ 
  • La syntax de puntos se puede combinar con caracteres comodín ( * ) para seleccionar (posiblemente múltiples) campos sin especificar nombres explícitamente:

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

Una vez que lo convierte a DF, puede simplemente obtener datos como

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

aquí fue lo que hice y funcionó

 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 

Resultados

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