Consulta ao Spark DataFrame SQL com tipos complexos

Como posso consultar um RDD com tipos complexos, como mapas / matrizes? Por exemplo, quando eu estava escrevendo este código de teste:

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

Eu pensei que a syntax seria algo como:

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

ou

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

mas eu fico

Não é possível acessar o campo nested no tipo MapType (StringType, StringType, true)

e

org.apache.spark.sql.catalyst.errors.package $ TreeNodeException: Atributos não resolvidos

respectivamente.

Depende de um tipo da coluna. Vamos começar com alguns dados fictícios:

 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) 
  • colunas da matriz:

    • Método Column.getItem

       df.select($"an_array".getItem(1)).show // +-----------+ // |an_array[1]| // +-----------+ // | 2| // | 5| // +-----------+ 
    • Sintaxe entre colchetes:

       sqlContext.sql("SELECT an_array[1] FROM df").show // +---+ // |_c0| // +---+ // | 2| // | 5| // +---+ 
    • um 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| // +---------------+ 
  • colunas de mapa

    • Usando o método Column.getField :

       df.select($"a_map".getField("foo")).show // +----------+ // |a_map[foo]| // +----------+ // | bar| // | null| // +----------+ 
    • usando a syntax de colchetes do Hive:

       sqlContext.sql("SELECT a_map['foz'] FROM df").show // +----+ // | _c0| // +----+ // |null| // | baz| // +----+ 
    • usando um caminho completo com syntax de ponto:

       df.select($"a_map.foo").show // +----+ // | foo| // +----+ // | bar| // |null| // +----+ 
    • usando um 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| // +--------------+ 
  • colunas de estrutura usando caminho completo com syntax de ponto:

    • com a API DataFrame

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

       sqlContext.sql("SELECT a_struct.x FROM df").show // +---+ // | x| // +---+ // | 1| // | 2| // +---+ 
  • campos dentro da matriz de structs podem ser acessados ​​usando a syntax de pontos, nomes e methods de Column padrão:

     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| // +------------------------------+ 
  • campos de tipos definidos pelo usuário (UDTs) podem ser acessados ​​usando UDFs. Consulte os atributos de referência do SparkSQL da UDT para obter detalhes.

Notas :

  • dependendo de uma versão do Spark, alguns desses methods podem estar disponíveis apenas com o HiveContext . As UDFs devem funcionar independentemente da versão com o SQLContext e o SQLContext padrão.
  • De um modo geral, os valores nesteds são cidadãos de segunda class. Nem todas as operações típicas são suportadas em campos nesteds. Dependendo de um contexto, pode ser melhor nivelar o esquema e / ou explodir collections

     df.select(explode($"an_array_of_structs")).show // +--------------------+ // | col| // +--------------------+ // |[foo,1,WrappedArr...| // |[bar,2,WrappedArr...| // |[foz,3,WrappedArr...| // |[baz,4,WrappedArr...| // +--------------------+ 
  • A syntax de ponto pode ser combinada com um caractere curinga ( * ) para selecionar (possivelmente vários) campos sem especificar nomes explicitamente:

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

Depois de convertê-lo para DF, você pode simplesmente buscar dados 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") 

aqui foi o que eu fiz e funcionou

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