Wie exportiere ich einen Tabellendatenrahmen in PySpark nach CSV?

76

Ich verwende Spark 1.3.1 (PySpark) und habe mithilfe einer SQL-Abfrage eine Tabelle generiert. Ich habe jetzt ein Objekt, das a ist DataFrame. Ich möchte dieses DataFrameObjekt (ich habe es "Tabelle" genannt) in eine CSV-Datei exportieren, damit ich es bearbeiten und die Spalten zeichnen kann. Wie exportiere ich die DataFrame"Tabelle" in eine CSV-Datei?

Vielen Dank!

python apache-spark dataframe apache-spark-sql export-to-csv PyRsquared
quelle

Antworten:

185

Wenn der Datenrahmen in einen Treiberspeicher passt und Sie ihn in einem lokalen Dateisystem speichern möchten, können Sie Spark DataFrame mithilfe der folgenden toPandasMethode in einen lokalen Pandas DataFrame konvertieren und dann einfach Folgendes verwenden to_csv:

df.toPandas().to_csv('mycsv.csv')

Andernfalls können Sie spark-csv verwenden :

  • Funke 1.3

    df.save('mycsv.csv', 'com.databricks.spark.csv')

  • Spark 1.4+

    df.write.format('com.databricks.spark.csv').save('mycsv.csv')

In Spark 2.0+ können Sie die csvDatenquelle direkt verwenden:

df.write.csv('mycsv.csv')
null323
quelle
Super Antwort. Ist dies für die erste Option möglich, wenn ich in eine durch Pipe getrennte Datei anstelle einer durch Kommas getrennten CSV schreiben möchte?
Brian Waters
7
Wenn Sie Spark-Datenrahmen haben, die Sie verwenden können df.write.csv('/tmp/lookatme/'), wird eine Reihe von CSV-Dateien in /tmp/lookatmeUsing Spark gelöscht. Dies ist erheblich schneller als die Serialisierung in Pandas. Der einzige Nachteil ist, dass Sie am Ende eine Reihe von CSVs anstelle einer einzigen haben. Wenn das Ziel-Tool nicht weiß, wie man sie verkettet, müssen Sie dies selbst tun.
Txangel
1
Was für eine große Sache ist es, eine CSV aus dem Funken zu bekommen. Interessant an dieser ersten Lösung ist, dass sie to_csvfunktioniert, ohne dass Pandas importiert werden müssen. .toPandasist ein Teil von Spark, vielleicht importiert es es implizit ..
Kardamom
21
Sie sollten in der Lage sein, zu verwenden, df.coalesce(1).write.csv('mycsv.csv')wenn Sie darauf bestehen, eine einzige Ausgabedatei zu haben
MichaelChirico
1
@ TXangel danke für deine Antwort. Wenn ich das verwende, läuft es jedoch fehlerfrei, aber ich kann keine am Zielspeicherort erstellte CSV finden . Irgendwelche Gedanken?
Rotail
33

Für Apache Spark 2+, um den Datenrahmen in einer einzelnen CSV-Datei zu speichern. Verwenden Sie den folgenden Befehl

query.repartition(1).write.csv("cc_out.csv", sep='|')

Hier 1geben Sie an, dass ich nur eine Partition von csv benötige. Sie können es entsprechend Ihren Anforderungen ändern.

Shafiq
quelle
6
Wie hier angegeben: spark.apache.org/docs/2.2.0/api/python/… wird empfohlen, coalesce () anstelle von repartition () zu verwenden, um die Leistung zu steigern ("Wenn Sie die Anzahl der Partitionen in diesem RDD verringern." , erwägen Sie die Verwendung von Koaleszenz, um ein Mischen zu vermeiden. ")
Seastar
@Seastar: Während das Zusammenführen in mehreren Anwendungsfällen Vorteile haben kann, gilt Ihr Kommentar in diesem speziellen Fall nicht. Wenn Sie eine CSV-Datei in Ihrem HDFS haben möchten (oder was auch immer), möchten Sie normalerweise eine Datei und nicht Dutzende von Dateien, die über Ihren Cluster verteilt sind (der ganze Sinn des Tuns repartition(1). Sie müssen die Daten dafür in beide Richtungen mischen Das Zusammenwachsen wird im Großen und Ganzen überhaupt nicht helfen
Markus,
19

Wenn Sie spark-csv nicht verwenden können, können Sie Folgendes tun:

df.rdd.map(lambda x: ",".join(map(str, x))).coalesce(1).saveAsTextFile("file.csv")

Wenn Sie Zeichenfolgen mit Zeilenumbrüchen oder Komma behandeln müssen, funktioniert dies nicht. Benutze das:

import csv
import cStringIO

def row2csv(row):
    buffer = cStringIO.StringIO()
    writer = csv.writer(buffer)
    writer.writerow([str(s).encode("utf-8") for s in row])
    buffer.seek(0)
    return buffer.read().strip()

df.rdd.map(row2csv).coalesce(1).saveAsTextFile("file.csv")
jbochi
quelle
7

Sie müssen den Datenrahmen in einer einzelnen Partition neu partitionieren und dann das Format, den Pfad und andere Parameter für die Datei im Unix-Dateisystemformat definieren.

df.repartition(1).write.format('com.databricks.spark.csv').save("/path/to/file/myfile.csv",header = 'true')

Lesen Sie mehr über die Repartitionsfunktion Lesen Sie mehr über die Speicherfunktion

Die Neupartitionierung ist jedoch eine kostspielige Funktion und toPandas () ist am schlechtesten. Verwenden Sie in der vorherigen Syntax .coalesce (1) anstelle von .repartition (1), um eine bessere Leistung zu erzielen.

Lesen Sie mehr über Repartition vs Coalesce-Funktionen .

Gazal Patel
quelle
2

Wie wäre es damit (wenn Sie keinen Einzeiler wollen)? 

for row in df.collect():
    d = row.asDict()
    s = "%d\t%s\t%s\n" % (d["int_column"], d["string_column"], d["string_column"])
    f.write(s)

f ist ein geöffneter Dateideskriptor. Auch das Trennzeichen ist ein TAB-Zeichen, aber es ist einfach, zu ändern, was Sie wollen.

Matei Florescu
quelle