Funke: schreiben DataFrame als komprimierte JSON

Apache Spark DataFrameReader.json() verarbeiten kann gzipped JSONlines Dateien automatisch, aber es scheint nicht zu einem Weg, um DataFrameWriter.json() zu schreiben komprimiert JSONlines-Dateien. Die zusätzliche Netzwerk-I/O ist sehr teuer in der cloud.

Gibt es einen Weg um dieses problem herum?

  • Haben Sie entdecken einen Weg, der zum komprimieren json-Ausgabe? Ich bin auch auf der Suche nach einer Lösung.
  • Ich habe nicht einen Weg fand, das zu tun dieses noch.
InformationsquelleAutor Sim | 2015-08-11
  1. 13

    Folgende Lösungen verwenden pyspark, aber ich nehme an, den code in Scala wäre ähnlich.

    Erste option ist, um die folgenden, wenn Sie initialisieren Ihre SparkConf:

    conf = SparkConf()
conf.set("spark.hadoop.mapred.output.compress", "true")
conf.set("spark.hadoop.mapred.output.compression.codec", "org.apache.hadoop.io.compress.GzipCodec")
conf.set("spark.hadoop.mapred.output.compression.type", "BLOCK")

    Mit dem code über jede Datei, die Sie produzieren mit, dass sparkContext automatisch mit gzip komprimiert.

    Zweite option, wenn Sie möchten, zu komprimieren ausgewählten Dateien werden nur in Ihrem Kontext. Können sagen, "df" ist ein dataframe und mit dem Namen Ihres Ziels:

    df_rdd = self.df.toJSON() 
df_rdd.saveAsTextFile(filename,compressionCodecClass="org.apache.hadoop.io.compress.GzipCodec")
    • Die Scala RDD-API ist def saveAsTextFile(path: String, codec: Class[_ <: CompressionCodec]) so, dass der code der Klasse sollte direkt übergeben und nicht als string.
    • Frage mich, ob es möglich ist, zu vermeiden, die hadoopish-format beim speichern der Daten in eine Datei. Ich kann nicht verwenden das Verzeichnis mit _SUCCES und part-* - Datei. Ich brauche nur einen spezifischen Namen eine einzige Datei...
    • Sorry für die Auferstehung, aber ich finde es schwer zu glauben, dass conf.set("spark.hadoop.mapred.output.compression.codec", "true") notwendig ist
    • DataFrame ist nicht RDD. Ändern Sie die Kompression Einstellungen Global, so dass es implizit ist KEINE gute Praxis als gut.
    InformationsquelleAutor giorgioca
  2. 23

    Mit Spark 2.X (und vielleicht auch früher, habe ich nicht getestet) es gibt eine einfachere Art und Weise zu schreiben ein komprimiertes JSON, die keine änderung der Konfiguration:

    val df: DataFrame = ...
df.write.option("compression", "gzip").json("/foo/bar")

    Dies funktioniert auch für CSV und Parkett, verwenden Sie einfach .csv() und .Parkett() statt .json() zu schreiben, die Datei nach der Einstellung der compression-option.

    Möglichen codecs: keine, bzip2, deflate, gzip, lz4 und bissig.

    • Es scheint nicht zu funktionieren mit früheren Versionen als Spark 2.X
    InformationsquelleAutor nsantos
  3. 6

    Einstellung der compression-Option auf einen SparkConf ist NICHT eine gute Praxis, da die akzeptierte Antwort. Es verändert das Verhalten Global statt der Angabe der Einstellungen auf einer pro-Datei-basis. Die Wahrheit ist, explizit ist immer besser als implizit. Es gibt auch einige Fälle, in denen der Benutzer nicht manipulieren kann, die Kontext-Konfiguration einfach, wie Funke-shell oder in codes entwickelt, die als Submodul von anderen.

    Der richtige Weg

    Schreiben DataFrame mit Kompression unterstützt seit Spark 1.4. Mehrere Möglichkeiten, um zu erreichen, dass:

    Eine

    df.write.json("filename.json", compression="gzip")

    Das ist es! Verwenden Sie einfach DataFrameWriter.json() wie Sie es wünschen.

    Ist die Magie versteckt in den code pyspark/sql/readwriter.py

    @since(1.4)
def json(self, path, mode=None, compression=None, dateFormat=None, timestampFormat=None):
    """Saves the content of the :class:`DataFrame` in JSON format
    (`JSON Lines text format or newline-delimited JSON <http://jsonlines.org/>`_) at the
    specified path.

    :param path: the path in any Hadoop supported file system
    :param mode: ...

    :param compression: compression codec to use when saving to file. This can be one of the
                        known case-insensitive shorten names (none, bzip2, gzip, lz4,
                        snappy and deflate).
    :param dateFormat: ...
    :param timestampFormat: ...

    >>> df.write.json(os.path.join(tempfile.mkdtemp(), 'data'))
    """
    self.mode(mode)
    self._set_opts(
        compression=compression, dateFormat=dateFormat, timestampFormat=timestampFormat)
    self._jwrite.json(path)

    Unterstützt Kompression Formate bzip2, gzip, lz4, bissig und entlüften, groß-und Kleinschreibung.

    Der scala-API sollte das gleiche sein.

    Anderen

    df.write.options(compression="gzip").json("filename.json")

    Ähnlich wie oben. weitere Optionen können suppied als Schlüsselwort-Argumente. seit Spark 1.4.

    Dritten

    df.write.option("compression", "gzip").json("filename.json")

    DataFrameWriter.option() ist, Hinzugefügt, da Spark 1.5. Nur ein parameter Hinzugefügt werden konnte zu einer Zeit.

    InformationsquelleAutor ttimasdf
Schreibe einen Kommentar