Programmgesteuert generieren, die das schema UND die Daten für einen dataframe in Apache Spark

Möchte ich dynamisch generieren Sie einen dataframe mit einem Kopf-Datensatz für einen Bericht so erstellen Sie einen dataframe aus dem Wert die Zeichenfolge unten:

val headerDescs : String = "Name,Age,Location"

val headerSchema = StructType(headerDescs.split(",").map(fieldName => StructField(fieldName, StringType, true)))

Aber jetzt möchte ich das gleiche tun für die Daten (die praktisch die gleichen Daten, d.h. der Metadaten).

Erstelle ich ein RDD :

val headerRDD = sc.parallelize(headerDescs.split(","))

Ich dann verwenden soll createDataFrame, um es zu schaffen:

val headerDf = sqlContext.createDataFrame(headerRDD, headerSchema)

jedoch fehl, weil createDataframe erwartet ein RDD[Row] aber meine RDD ist ein array von strings - ich kann nicht finden, eine Möglichkeit der Umwandlung meiner RDD auf eine Zeile RDD und dann das mapping der Felder dynamisch. Beispiele, die ich gesehen habe, davon ausgehen, Sie kennen die Anzahl von Spalten, aber vorher will ich die Möglichkeit irgendwann in der Lage sein zu ändern die Spalten ohne änderung der code - die Spalten in eine Datei zum Beispiel.

Code-Auszug auf der Grundlage der ersten Antwort:

val headerDescs : String = "Name,Age,Location"

//create the schema from a string, splitting by delimiter
val headerSchema = StructType(headerDescs.split(",").map(fieldName => StructField(fieldName, StringType, true)))

//create a row from a string, splitting by delimiter
val headerRDDRows = sc.parallelize(headerDescs.split(",")).map( a => Row(a))

val headerDf = sqlContext.createDataFrame(headerRDDRows, headerSchema)
headerDf.show()

Ausführen dieser Ergebnisse, in:

+--------+---+--------+

|    Name|Age|Location|

+--------+---+--------+

|    Name|

|     Age|

|Location|

+--------+---+-------

InformationsquelleAutor Jon Robinson | 2017-01-19

3

Für die Umwandlung RDD[Array[String]] zu RDD[Row] müssen Sie die folgenden Schritte aus:

import org.apache.spark.sql.Row
```
val headerRDD = sc.parallelize(Seq(headerDescs.split(","))).map(x=>Row(x(0),x(1),x(2)))

scala> val headerSchema = StructType(headerDescs.split(",").map(fieldName => StructField(fieldName, StringType, true)))
headerSchema: org.apache.spark.sql.types.StructType = StructType(StructField(Name,StringType,true), StructField(Age,StringType,true), StructField(Location,StringType,true))

scala> val headerRDD = sc.parallelize(Seq(headerDescs.split(","))).map(x=>Row(x(0),x(1),x(2)))
headerRDD: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[6] at map at <console>:34

scala> val headerDf = sqlContext.createDataFrame(headerRDD, headerSchema)
headerDf: org.apache.spark.sql.DataFrame = [Name: string, Age: string, Location: string]


scala> headerDf.printSchema
root
 |-- Name: string (nullable = true)
 |-- Age: string (nullable = true)
 |-- Location: string (nullable = true)



scala> headerDf.show
+----+---+--------+
|Name|Age|Location|
+----+---+--------+
|Name|Age|Location|
+----+---+--------+
```
Diese geben Ihnen eine RDD[Row]

Zum Lesen über Datei
```
val vRDD = sc.textFile("..**filepath**.").map(_.split(",")).map(a => Row.fromSeq(a))

val headerDf = sqlContext.createDataFrame(vRDD , headerSchema)
```
Mit Spark-CSV Paket :
```
 val df = sqlContext.read
    .format("com.databricks.spark.csv")
    .option("header", "true") //Use first line of all files as header
    .schema(headerSchema) //defining based on the custom schema
    .load("cars.csv")
```
ODER
```
val df = sqlContext.read
    .format("com.databricks.spark.csv")
    .option("header", "true") //Use first line of all files as header
    .option("inferSchema", "true") //Automatically infer data types
    .load("cars.csv")
```
Gibt es verschiedene Optionen, die Sie erkunden können, in seiner Dokumentation.
- Vielen Dank für Ihre schnelle Antwort, aber ich bekomme die Fehlermeldung: 45: Fehler: Wert geteilt wird, der nicht Mitglied der Char
- aktualisiert haben die Antwort .sollte dies funktionieren
- Danke, das ist näher, aber abgebildet hat alle Werte auf den ersten dataframe Spalte, d.h. alle Werte unter der Spalte "Name" ich will "Name" unter der Spalte Name, 'Alter' unter der Spalte Alter usw.
- können Sie die Ausgabe, die Sie bekommen und den code.
- denn wenn man sieht, sogar das schema zeigt 3 Spalten
- Das schema ist in Ordnung, das problem ist die Daten angezeigt, die unter der ersten Spalte, also 'Name', 'Alter', 'Ort' erscheinen unter der Spalte Name.
- bitte aus ur scheint es u erstellen von leeren dataframe. poste bitte den code, wo u r einfügen von Daten in dataframe.
- Aktualisiert die ursprüngliche Frage mit dem code und Ausgabe.
- Danke für den Vorschlag, aber ich hatte bereits reduzierte diesen Ansatz erneut das Problem ist, dass Sie brauchen, um zu wissen, wie viele Spalten in der Ausgabe vorher, d.h. es ist nicht programmatisch. Ich wollte es so, dass wenn du ein update headerDescs z.B. headerDescs = "Name, Alter, Wohnort, E-Mail", die Sie nicht haben, die andere zu ändern code.
- von wo liest du die Daten . Ich meine, ist es aus einer Datei oder Tabelle ?
- letztlich wird es aus einer Datei gelesen.
- wenn Sie beim Lesen aus einer Datei und Umwandlung in einen dataframe, ich schlage vor, Sie verwenden spark-csv-Paket. github.com/databricks/spark-csv
- aktualisiert den Antwort erstellen dataframe aus einer Datei.
- Ich mag deine Antwort für das Lesen aus einer Datei, das ist das, was ich Suche. Vielen Dank für deine Hilfe, Rajat.
InformationsquelleAutor Rajat Mishra

Schreibe einen Kommentar

Du musst angemeldet sein, um einen Kommentar abzugeben.