Schreiben Spark dataframe als CSV mit Partitionen

Ich versuche zu schreiben dataframe in spark zu einem HDFS-Lage, und ich erwarte, dass wenn ich hinzufüge, die partitionBy notation Funke partition erstellen
(ähnlich wie beim schreiben in Parkett-format)
Ordner in form von 

partition_column_name=partition_value

( ich.e partition_date=2016-05-03). Dazu habe ich folgenden Befehl ausgeführt :

(df.write
    .partitionBy('partition_date')
    .mode('overwrite')
    .format("com.databricks.spark.csv")
    .save('/tmp/af_organic'))

aber partition Ordner wurde nicht erstellt
jede Idee, was könnte ich tun, um für die spark-DF automatisch erstellen diese Ordner?

Dank,

InformationsquelleAutor Lior Baber | 2016-05-29
  1. 20

    Funke 2.0.0+:

    Gebaut-in das csv-format unterstützt die Partitionierung aus der box, so dass Sie sollten in der Lage sein verwenden Sie einfach:

    df.write.partitionBy('partition_date').mode(mode).format("csv").save(path)

    ohne zusätzliche Pakete.

    Funke < 2.0.0:

    In diesem moment (v1.4.0) spark-csv nicht unterstützen partitionBy (siehe databricks/Funke-csv#123), aber Sie können anpassen, built-in Quellen zu erreichen, was Sie wollen.

    Können Sie versuchen, zwei unterschiedliche Ansätze. Vorausgesetzt, Ihre Daten relativ einfach ist (keine komplexe Zeichenfolgen und müssen für Zeichen-escaping) und sieht mehr oder weniger wie diese:

    df = sc.parallelize([
    ("foo", 1, 2.0, 4.0), ("bar", -1, 3.5, -0.1)
]).toDF(["k", "x1", "x2", "x3"])

    Können Sie manuell vorzubereiten Werte für das schreiben:

    from pyspark.sql.functions import col, concat_ws

key = col("k")
values = concat_ws(",", *[col(x) for x in df.columns[1:]])

kvs = df.select(key, values)

    schreiben und mit text Quelle

    kvs.write.partitionBy("k").text("/tmp/foo")

df_foo = (sqlContext.read.format("com.databricks.spark.csv")
    .options(inferSchema="true")
    .load("/tmp/foo/k=foo"))

df_foo.printSchema()
## root
## |-- C0: integer (nullable = true)
## |-- C1: double (nullable = true)
## |-- C2: double (nullable = true)

    In komplexeren Fällen können Sie versuchen, um die ordnungsgemäße Verwendung von CSV-parser Vorverarbeiten Werte in einer ähnlichen Weise, entweder durch Verwendung von UDF-oder mapping über RDD, aber es wird deutlich teurer.

    Wenn das CSV-format ist nicht eine harte Anforderung können Sie auch JSON verwenden writer unterstützt partitionBy out-of-the-box:

    df.write.partitionBy("k").json("/tmp/bar")

    sowie partition Entdeckung auf Lesen.

    InformationsquelleAutor zero323
Schreibe einen Kommentar