如何将 DataFrame 保存为压缩（gzipped）CSV？

Question

我使用 Spark 1.6.0 和 Scala。

我想将 DataFrame 保存为压缩的 CSV 格式。

这是我目前所拥有的（假设我已经拥有 df 和 sc 作为 SparkContext）：

//set the conf to the codec I want
sc.getConf.set("spark.hadoop.mapred.output.compress", "true")
sc.getConf.set("spark.hadoop.mapred.output.compression.codec", "true")
sc.getConf.set("spark.hadoop.mapred.output.compression.codec", "org.apache.hadoop.io.compress.GzipCodec")
sc.getConf.set("spark.hadoop.mapred.output.compression.type", "BLOCK")

df.write
  .format("com.databricks.spark.csv")
  .save(my_directory)

输出不是 gz 格式。

Answer 1

Spark 2.2+

df.write.option("compression","gzip").csv("path")

Spark 2.0

df.write.csv("path", compression="gzip")

Spark 1.6

在 spark-csv 上 github： https://github.com/databricks/spark-csv

一个人可以阅读：

codec: compression codec to use when saving to file. Should be the fully qualified name of a class implementing org.apache.hadoop.io.compress.CompressionCodec or one of case-insensitive shorten names (bzip2, gzip, lz4, and snappy). Defaults to no compression when a codec is not specified.

在这种情况下，这是有效的： df.write.format("com.databricks.spark.csv").codec("gzip")\ .save('my_directory/my_file.gzip')

Answer 2

此代码适用于 Spark 2.1，其中 .codec 不可用。

df.write
  .format("com.databricks.spark.csv")
  .option("codec", "org.apache.hadoop.io.compress.GzipCodec")
  .save(my_directory)

对于 Spark 2.2，您可以使用此处描述的 df.write.csv(...,codec="gzip") 选项：https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=codec

Answer 3

有了 Spark 2.0+，这变得更简单了：

df.write.csv("path", compression="gzip")  # Python-only
df.write.option("compression", "gzip").csv("path") // Scala or Python

您不再需要外部 Databricks CSV 包。

csv() 编写器支持许多方便的选项。例如：

sep: 设置分隔符。
quote：是否以及如何引用值。
header：是否包含header行。

除了 gzip:

之外，您还可以使用许多其他压缩编解码器

bzip2
lz4
snappy
deflate

csv() 作者的完整 Spark 文档在此处：Python / Scala

Answer 4

用 headers 写入 CSV 文件并将 part-000 文件重命名为 .csv.gzip

DF.coalesce(1).write.format("com.databricks.spark.csv").mode("overwrite")
.option("header","true")
.option("codec",org.apache.hadoop.io.compress.GzipCodec").save(tempLocationFileName)

copyRename(tempLocationFileName, finalLocationFileName)

def copyRename(srcPath: String, dstPath: String): Unit =  {
  val hadoopConfig = new Configuration()
  val hdfs = FileSystem.get(hadoopConfig)
  FileUtil.copyMerge(hdfs, new Path(srcPath), hdfs, new Path(dstPath), true, hadoopConfig, null)
  // the "true" setting deletes the source files once they are merged into the new output
}

如果您不需要 header，则将其设置为 false，您也不需要进行合并。写起来也会快一些。

如何将 DataFrame 保存为压缩（gzipped）CSV？

How to save a DataFrame as compressed (gzipped) CSV?

csv

scala

apache-spark

spark-dataframe