Wie schreibe ich CSV-Datei in eine Datei von pyspark

Question

Ich verwende this Methode, um CSV-Datei schreiben. Aber es wird eine Datei mit mehreren Teiledateien erzeugen. Das ist nicht was ich will; Ich brauche es in einer Datei. Und ich fand auch another post mit scala, um alles zu erzwingen, auf einer Partition berechnet zu werden, und dann eine Datei zu bekommen.

Erste Frage: Wie erreicht man das in Python?

Im zweiten Post wird auch gesagt, dass ein Hadoop function mehrere Dateien zu einem zusammenführen könnte.

Zweite Frage: ist es möglich, zwei Dateien in Spark zusammenführen?

Answer 1

Sie können dies tun, indem Sie die cat-Befehlszeilenfunktion wie unten verwenden. Dies wird alle Teildateien in 1 csv verketten. Es ist nicht erforderlich, bis zu 1 Partition neu zu partitionieren.

Answer 2

Die Anforderung besteht darin, eine RDD in einer einzigen CSV-Datei zu speichern, indem die RDD an einen Executor übergeben wird. Dies bedeutet, dass RDD-Partitionen, die auf mehreren Executoren vorhanden sind, zu einem Executor gemischt werden. Wir können coalesce(1) oder repartition(1) für diesen Zweck verwenden. Zusätzlich dazu kann man der resultierenden csv-Datei eine Spaltenüberschrift hinzufügen. Zuerst können wir eine Dienstprogrammfunktion für make Daten csv kompatibel halten.

def toCSVLine(data): 
    return ','.join(str(d) for d in data) 

Nehmen wir an, MyRDD fünf Spalten hat, und es muss 'ID', 'DT_KEY', 'Grade', 'Score', 'TRF_Age' als Spaltenüberschriften. So erstelle ich eine Kopf-RDD und Union MyRDD wie unten, die die meisten Male hält die Kopfzeile über der CSV-Datei.

unionHeaderRDD = sc.parallelize([('ID','DT_KEY','Grade','Score','TRF_Age')])\ 
    .union(MyRDD) 

unionHeaderRDD.coalesce(1).map(toCSVLine).saveAsTextFile("MyFileLocation") 

saveAsPickleFile Funken Kontext API-Methode kann verwendet werden, um Daten zu serialisieren, die um Platz sparend gespeichert wird. Verwenden Sie pickFile, um die eingelegte Datei zu lesen.

Answer 3

Sie verwenden können,

df.coalesce (1) .write.csv ('Result.csv')

Aber wenn man coalesce Funktion verwenden, werden Sie Ihre Parallelität verlieren.