Запрос Spark SQL DataFrame со сложными типами

Как я могу запросить 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)), 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) 
  • столбцы массива:

    • Метод Column.getItem

       df.select($"an_array".getItem(1)).show // +-----------+ // |an_array[1]| // +-----------+ // | 2| // | 5| // +-----------+ 
    • Синтаксис скобок улья:

       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| // +---------------+ 
  • столбцы карт

    • используя метод 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| // +--------------+ 
  • Структурные столбцы, используя полный путь с точечным синтаксисом:

    • с API DataFrame

       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. Для получения дополнительной информации см. SparkSQL, ссылающиеся на атрибуты UDT .

Примечания :

  • в зависимости от версии 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...| // +--------------------+ 
  • Синтаксис Dot можно комбинировать с подстановочным символом ( * ) для выбора (возможно, нескольких) полей без явного указания имен:

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

После того, как вы конвертируете его в 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") 

вот что я сделал, и это сработало

 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 

Результаты

 +------+ | hello| +------+ | world| |people| +------+ 
  • Столбец Access Array в Spark
  • Каковы возможные причины для получения TimeoutException: фьючерсы, истекающие после при работе с Spark
  • Как импортировать несколько файлов csv в одной загрузке?
  • Как передать дополнительные параметры UDF в SparkSql?
  • Оптимизация соединения DataFrame - Broadcast Hash Join
  • Как получить доступ к элементу столбца VectorUDT в Spark DataFrame?
  • Как запросить столбцы данных JSON, используя Spark DataFrames?
  • Spark Sql UDF со сложным входным параметром
  • Перезаписывать конкретные разделы в методе записи данных в режиме искривления
  • Как найти размер вспышки RDD / Dataframe?
  • Давайте будем гением компьютера.