Wie zu veröffentlichen einzelne Parkett-Datei nach der Verarbeitung in Spark

Question

Ich habe eine Parkett-Datei mit 100.000 Datensätze. Ich möchte alle Datensätze mit Parallelität verarbeiten und meine Verarbeitung wird weitere Spalten generieren. Also, wenn Ihr Parkett Datei 3 Spalten und 100.000 Datensätze hat, sieht meine Datei wie wie unten-

col A col B  colC 
aa   bb  cc 
aa1  bb1  cc1 

Nachdem in parallel verarbeitet zu werden, habe ich eine neue Parkett Datei mit gleichen Spalten und 3more Spalten möchten. Meine Ausgabe sieht wie folgt aus -

col A col B  colC  colD  colE  colF 
aa   bb  cc  dd  ee   ff 
aa1  bb1  cc1  dd1  ee1   ff1 

Ich möchte wissen -

• Nachdem parallel in Funken Knoten ausgeführt wird, wie kann ich alle Ergebnisse in 1 Parkett-Datei kombinieren?
• Wie kann ich mehr Spalten zu einer vorhandenen Datei hinzufügen?

Jede Hilfe wäre willkommen.

Answer 1

Sie können jederzeit Ihre Daten auf eine einzelne Partition neu partitionieren mit

df.repartition(1) 

Es wird die Arbeit tun, wenn Sie nur das Experimentieren sind, oder die Arbeit mit kleinen Datenmengen (100.000 Datensatz mit hinreichender Länge können kleine Daten betrachtet werden) . Aber wenn Sie Ihre Anwendung auf die Waage bringen wollen, sollten Sie eine solch aggressive Neupartitionierung vermeiden und mehrere Parkett-Dateien schreiben. Was ist so schlimm daran, mehrere Ausgabedateien zu haben?

In Bezug auf Ihre zweite Frage, Parkett-Dateien entwickelt, um unveränderlich zu sein, so die einfachste Art zu tun, was Sie wollen, wird das Lesen der Daten sein, fügen Sie die relevanten Spalten und speichern Sie die Daten zurück als Parkett oder speichern Sie die geänderten Daten in einem neuen Verzeichnis).

Answer 2

In diesem Fall empfehle ich Ihnen, keine einzige Datei von Apache Spark zu erstellen, da dies die Leistung des Tools beeinträchtigt. Wenn Sie die Lagerung in einem HDFS tun empfehle ich Ihnen, diesen Befehl auszuführen, nachdem Sie Ihre verkleinern beenden:

hadoop fs -getmerge /output/dir/on/hdfs/ /desired/local/output/file.txt 

Oder Sie können eine Funktion übernehmen, der diese Dateien für Sie wie folgt zusammenführen können:

textFiles = [f for f in listdir(outputPath) if isfile(join(outputPath, f))] 

files = [f for f in textFiles if f.startswith('part-')] 
files.sort() 

filePath = outputPath + '/consolidate_file' 
outputFile = open(filePath, 'w') 

for i in files: 
    tmpFile = open(outputPath + '/' + i, 'r') 
    outputFile.write(tmpFile.read()) 

outputFile.close()