Apache Spark的DataFrameReader.json()
可以自动处理压缩的JSONlines文件,但似乎没有办法让DataFrameWriter.json()
写入压缩的JSONlines文件。在云中进行额外的网络I/O非常昂贵。
有没有解决这个问题的方法?
Apache Spark的DataFrameReader.json()
可以自动处理压缩的JSONlines文件,但似乎没有办法让DataFrameWriter.json()
写入压缩的JSONlines文件。在云中进行额外的网络I/O非常昂贵。
有没有解决这个问题的方法?
使用Spark 2.X(以及可能更早版本,我没有测试)有一种更简单的方法来编写压缩JSON,而不需要改变配置:
val df: DataFrame = ...
df.write.option("compression", "gzip").json("/foo/bar")
这也适用于CSV和Parquet,只需在设置压缩选项后使用.csv()和.parquet()代替.json()来写入文件。
可能的编解码器有:none、bzip2、deflate、gzip、lz4和snappy。
设置SparkConf
的压缩选项不是一个好的做法,就像被接受的答案所说。它改变了行为全局而不是在每个文件上指示设置。事实是,显式总比隐式要好。还有一些情况下,用户无法轻易地操作上下文配置,例如spark-shell或作为另一个子模块设计的代码。
从Spark 1.4开始支持使用压缩编写DataFrame
。有多种方法可以实现:
df.write.json("filename.json", compression="gzip")
就这样!只需随心所欲地使用DataFrameWriter.json()
。
魔法隐藏在代码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)
支持的压缩格式包括bzip2,gzip,lz4,snappy和deflate,不区分大小写。
Scala API应该是相同的。
df.write.options(compression="gzip").json("filename.json")
与上面类似。更多选项可以作为关键字参数提供。自 Spark 1.4 以来可用。
df.write.option("compression", "gzip").json("filename.json")
DataFrameWriter.option()
自Spark 1.5版本以后加入。每次只能添加一个参数。
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")
使用上面的代码,您使用该sparkContext生成的任何文件都将自动使用gzip压缩。
第二个选项是,如果您只想在上下文中压缩选定的文件。假设“df”是您的数据帧,而“filename”是您的目标:
df_rdd = self.df.toJSON()
df_rdd.saveAsTextFile(filename,compressionCodecClass="org.apache.hadoop.io.compress.GzipCodec")
def saveAsTextFile(path: String, codec: Class[_ <: CompressionCodec])
,因此代码类应该直接传递,而不是作为字符串传递。 - Simconf.set("spark.hadoop.mapred.output.compression.codec", "true")
是必要的。 - oarsome