Comment ajouter une nouvelle colonne Struct à un DataFrame

Je suis en train d'essayer d'extraire d'une base de données MongoDB et l'utilisation de l'Étincelle à l'acquérir dans ElasticSearch avec geo_points.

La base de données Mongo a des valeurs de latitude et longitude, mais ElasticSearch oblige à être intégré dans le geo_point type.

Est-il un moyen de l'Étincelle à la copie de la lat et lon colonnes à la une nouvelle colonne qui est une array ou struct?

Toute aide est très appréciée!

source d'informationauteur Kim Ngo

  1. 37

    Je suppose que vous commencez avec une sorte de plat de schéma comme ceci:

    root
 |-- lat: double (nullable = false)
 |-- long: double (nullable = false)
 |-- key: string (nullable = false)

    D'abord, permet de créer les données de l'exemple:

    import org.apache.spark.sql.Row
import org.apache.spark.sql.functions.{col, udf}
import org.apache.spark.sql.types._

val rdd = sc.parallelize(
    Row(52.23, 21.01, "Warsaw") :: Row(42.30, 9.15, "Corte") :: Nil)

val schema = StructType(
    StructField("lat", DoubleType, false) ::
    StructField("long", DoubleType, false) ::
    StructField("key", StringType, false) ::Nil)

val df = sqlContext.createDataFrame(rdd, schema)

    Un moyen facile est d'utiliser un udf et de la classe de cas:

    case class Location(lat: Double, long: Double)
val makeLocation = udf((lat: Double, long: Double) => Location(lat, long))

val dfRes = df.
   withColumn("location", makeLocation(col("lat"), col("long"))).
   drop("lat").
   drop("long")

dfRes.printSchema

    et nous obtenons 

    root
 |-- key: string (nullable = false)
 |-- location: struct (nullable = true)
 |    |-- lat: double (nullable = false)
 |    |-- long: double (nullable = false)

    Une voie difficile, c'est de transformer vos données et de les appliquer schéma par la suite:

    val rddRes = df.
    map{case Row(lat, long, key) => Row(key, Row(lat, long))}

val schemaRes = StructType(
    StructField("key", StringType, false) ::
    StructField("location", StructType(
        StructField("lat", DoubleType, false) ::
        StructField("long", DoubleType, false) :: Nil
    ), true) :: Nil 
)

sqlContext.createDataFrame(rddRes, schemaRes).show

    et nous obtenons un résultat attendu

    +------+-------------+
|   key|     location|
+------+-------------+
|Warsaw|[52.23,21.01]|
| Corte|  [42.3,9.15]|
+------+-------------+

    La création de imbriquée schéma à partir de zéro peut être fastidieux si vous le pouvez, je recommanderais de la première approche. Il peut être facilement étendu si vous avez besoin de plus sophistiqué de la structure:

    case class Pin(location: Location)
val makePin = udf((lat: Double, long: Double) => Pin(Location(lat, long))

df.
    withColumn("pin", makePin(col("lat"), col("long"))).
    drop("lat").
    drop("long").
    printSchema

    et nous vous prévu de sortie:

    root
 |-- key: string (nullable = false)
 |-- pin: struct (nullable = true)
 |    |-- location: struct (nullable = true)
 |    |    |-- lat: double (nullable = false)
 |    |    |-- long: double (nullable = false)

    Malheureusement, vous n'avez aucun contrôle sur nullable terrain, et donc il est important pour votre projet, vous devrez spécifier le schéma.

    Enfin, vous pouvez utiliser struct fonction introduite dans la version 1.4:

    import org.apache.spark.sql.functions.struct

df.select($"key", struct($"lat", $"long").alias("location"))
  2. 2

    Essayez ceci:

    import org.apache.spark.sql.functions._

df.registerTempTable("dt")

dfres = sql("select struct(lat,lon) as colName from dt")