Étincelle: l'écriture DataFrame comprimé JSON

Apache Spark est DataFrameReader.json() pouvez gérer au format gzip JSONlines automatiquement les fichiers, mais il ne semble pas être un moyen d'obtenir DataFrameWriter.json() à écrire comprimé JSONlines fichiers. Le réseau supplémentaire I/O est très cher dans le cloud.

Est-il un moyen de contourner ce problème?

Avez-vous trouver un moyen pour compresser les données json? Je suis aussi à la recherche d'une solution.
Je n'ai pas découvert le moyen de le faire encore.

OriginalL'auteur Sim | 2015-08-11

  1. 11

    Les solutions suivantes utilisation pyspark, mais je suppose que le code de la Scala serait similaire.

    Première option est de définir les options suivantes lorsque vous utilisez votre 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")

    Avec le code ci-dessus n'importe quel fichier que vous créez à l'aide de cette sparkContext est automatiquement compressé à l'aide de gzip.

    Deuxième option, si vous souhaitez compresser uniquement les fichiers sélectionnés à l'intérieur de votre cadre. Permet de dire "df" est votre dataframe et le nom de votre destination:

    df_rdd = self.df.toJSON() 
df_rdd.saveAsTextFile(filename,compressionCodecClass="org.apache.hadoop.io.compress.GzipCodec")
    La Scala RDD API est def saveAsTextFile(path: String, codec: Class[_ <: CompressionCodec]) de sorte que le code de la classe doit être transmis directement et non pas comme une chaîne de caractères.
    Vous vous demandez si c'est possible d'éviter les hadoopish format de stockage des données dans un fichier. Je ne peux pas utiliser le répertoire avec _SUCCES et part-* fichier. J'ai juste besoin d'un spécifique nommé seul fichier...
    Désolé au sujet de la résurrection, mais j'ai du mal à croire que conf.set("spark.hadoop.mapred.output.compression.codec", "true") est nécessaire
    DataFrame n'est pas RDD. Modifier les paramètres de compression au niveau mondial implicite n'est PAS une bonne pratique ainsi.

    OriginalL'auteur giorgioca

  2. 21

    Avec Spark 2.X (et peut-être plus tôt, je n'ai pas tester) il y a un moyen plus simple d'écrire un comprimé JSON, qui ne nécessite pas de modification de la configuration:

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

    Cela fonctionne aussi pour les CSV et pour le Parquet, il suffit d'utiliser .csv() et .parquet() au lieu de .json() pour écrire le fichier après le réglage de l'option de compression.

    La possible codecs sont: aucun, bzip2, dégonfler, gzip, lz4 et accrocheur.

    Il ne semble pas fonctionner avec les versions antérieures de Spark 2.X

    OriginalL'auteur nsantos

  3. 3

    Réglage des options de compression sur un SparkConf est PAS une bonne pratique, que l'on a accepté la réponse. Il a changé le comportement à l'échelle mondiale au lieu d'indiquer les paramètres de base de fichier par fichier. La vérité est, explicite est toujours mieux qu'implicites. Il y a aussi certains cas où les utilisateurs ne peuvent pas manipuler le contexte de configuration, comme l'étincelle-shell ou dans les codes conçu comme un sous-module d'un autre.

    La façon correcte de

    Écrit DataFrame avec la compression est prise en charge depuis l'Étincelle 1.4. Plusieurs manières de réaliser que:

    Un

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

    Que c'est! Utilisez simplement DataFrameWriter.json() comme vous le souhaitez.

    La magie est caché dans le 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)

    Pris en charge les formats de compression sont bzip2, gzip, lz4, accrocheur et dégonfler, insensible à la casse.

    La scala de l'API doit être le même.

    Un autre

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

    Similaires comme ci-dessus. plus d'options peut être suppied comme mot-clé arguments. disponible depuis Étincelle 1.4.

    Troisième

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

    DataFrameWriter.option() est ajouté depuis Étincelle 1.5. Un seul paramètre peut être ajouté à un moment.

    OriginalL'auteur ttimasdf