Hinzufügen eines neuen Struct-Spalte, um ein DataFrame

Ich bin gerade dabei, extrahieren Sie eine Datenbank von MongoDB und verwenden von Spark für das Einlesen in ElasticSearch mit geo_points.

Den Mongo-Datenbank hat, werden Längen-und Breitengrad-Werte, aber ElasticSearch verlangt, dass Sie gegossen werden in der geo_point geben.

Gibt es eine Möglichkeit in der Funke zu kopieren lat und lon Spalten zu einer neuen Spalte ist ein array oder struct?

Jede Hilfe ist willkommen!

InformationsquelleAutor Kim Ngo | 2015-07-24

  1. 46

    Ich nehme an, Sie beginnen mit irgendeiner Art von flat-schema so:

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

    Ersten lets-erstellen von Beispiel-Daten:

    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)

    Eine einfache Möglichkeit ist die Verwendung eines udf-und Fall-Klasse:

    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

    und wir bekommen 

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

    Ein harter Weg ist, um wandeln Sie Ihre Daten und wenden schema danach:

    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

    und erhalten wir die erwartete Ausgabe

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

    Erstellen von verschachtelten schema von Grund auf mühsam sein kann, so wenn Sie kann würde ich empfehlen den ersten Ansatz. Es kann leicht erweitert werden, wenn Sie mehr brauchen ausgefeilte Struktur:

    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

    und wir Holen Sie erwartete Ausgabe:

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

    Leider haben Sie keine Kontrolle über nullable Feld, so dass wenn ist wichtig für Ihr Projekt müssen Sie angeben schema.

    Schließlich können Sie struct Funktion eingeführt, die in 1.4:

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

df.select($"key", struct($"lat", $"long").alias("location"))
    Danke @zero323 für Ihre ausführliche Antwort! Dies hilft, einen Haufen. Würden Sie wissen, wie ich dies tun könnte mapping rekursiv geschachtelte Typen? Diese Daten hässlicher als ich gehofft hatte.
    Ich sehe keinen Grund, warum Sie es nicht konnte.
    Hi @zero323 - wissen Sie, wenn es trotzdem zu nutzen, ist Ihre UDF-Methode erstellen Sie eine struct, wenn es mehr als 10 Spalten in der neuen Struktur? UDFs zu haben scheinen, eine Beschränkung auf 10 input-Variablen.
    stackoverflow.com/q/39494620/1560062 helfen?
    Hi @zero323. Eigentlich dein letzter Vorschlag von "struct" - Funktion hat mir geholfen. Ich sollte Lesen Sie auf der Unterseite zuerst!

    InformationsquelleAutor zero323

  2. 3

    Versuchen Sie dies:

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

df.registerTempTable("dt")

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

    InformationsquelleAutor user8817325

Schreibe einen Kommentar