Rekursives Hinzufügen von Zeilen zu einem Datenrahmen

Question

Ich bin neu zu funken. Ich habe einige json Daten, die als HttpResponse kommt. Ich muss diese Daten in Hive-Tabellen speichern. Jede Anforderung HttpGet gibt einen JSON zurück, der eine einzelne Zeile in der Tabelle darstellt. Aus diesem Grund muss ich einzelne Zeilen als Dateien im Verzeichnis der Hive-Tabelle schreiben.

Aber ich fühle mich zu viele kleine Dateien wird die Geschwindigkeit und Effizienz zu reduzieren. Also gibt es eine Möglichkeit, dass ich rekursiv neue Zeilen zum Dataframe hinzufügen und es auf einmal in das Hive-Tabellenverzeichnis schreiben kann. Ich denke, das wird auch die Laufzeit meines Spark-Codes reduzieren.

Beispiel:

for(i <- 1 to 10){ 
newDF = hiveContext.read.json("path") 
df = df.union(newDF) 
} 
df.write() 

Ich verstehe, dass der Datenrahmen unveränderlich ist. Gibt es einen Weg, dies zu erreichen?

Jede Hilfe wäre willkommen. Vielen Dank.

Answer 1

Sie meist auf dem richtigen Weg sind, was Sie wollen, ist, mehrere einzelne Datensätze als Seq[DataFrame], und dann reduzieren Sie die Seq[DataFrame] zu einem einzigen DataFrame von unioning sie zu erhalten.

aus dem Code Gehen Sie zur Verfügung gestellt:

val BatchSize = 100 
val HiveTableName = "table" 

(0 until BatchSize). 
map(_ => hiveContext.read.json("path")). 
reduce(_ union _). 
write.insertInto(HiveTableName) 

Alternativ, wenn Sie die HTTP-Anfragen durchführen möchten, wie Sie gehen, können wir das auch tun. Nehmen wir an, Sie haben eine Funktion, die die HTTP-Anforderung hat und wandelt es in ein Datenrahmen:

def obtainRecord(...): DataFrame = ??? 

Sie können entlang der Linien von etwas tun:

val HiveTableName = "table" 
val OtherHiveTableName = "other_table" 
val jsonArray = ??? 

val batched: DataFrame = 
    jsonArray. 
    map { parameter => 
     obtainRecord(parameter) 
    }. 
    reduce(_ union _) 
batched.write.insertInto(HiveTableName) 
batched.select($"...").write.insertInto(OtherHiveTableName) 

Answer 2

Sie verwenden Spark eindeutig falsch. Apache Spark ist ein analytisches System, keine Datenbank-API. Es bietet keinen Vorteil, Spark zu verwenden, um die Hive-Datenbank so zu ändern. Dies kann nur zu einer erheblichen Leistungseinbuße führen, ohne von den Spark-Funktionen einschließlich der verteilten Verarbeitung zu profitieren.

Stattdessen sollten Sie den Hive-Client direkt für Transaktionsvorgänge verwenden.

Answer 3

Wenn Sie alle Daten im Batch-Download (zum Beispiel mit einem Skript mit curl oder einem anderen Programm) und speichern Sie sie in einer Datei (oder viele Dateien, Spark kann ein ganzes Verzeichnis auf einmal laden) können Sie Laden Sie dann diese Datei (oder Dateien) auf einmal in Spark, um Ihre Verarbeitung zu tun. Ich würde auch sehen, dass es die Webapi als Endpunkte, um alle Daten, die Sie benötigen, anstatt nur einen Datensatz auf einmal zu holen.