解压多个 *.gz 文件并在 spark scala 中制作一个 csv 文件

Posted 2023-04-17

【中文标题】解压多个 *.gz 文件并在 spark scala 中制作一个 csv 文件

【英文标题】：Unzip the multiple *.gz files and make one csv file in spark scala

【发布时间】：2018-09-26 15:34:24

【问题描述】：

我在 S3 存储桶中有多个文件，必须解压缩这些文件并将所有文件合并到一个带有单个标头的文件 (CSV) 中。所有文件都包含相同的标题。

数据文件如下所示。

存储系统：S3 存储桶。

 part-0000-XXXX.csv.gz
 part_0001-YYYY.csv.gz
 part-0002-ZZZZ.csv.gz
 .
 .
 .
 .
 part-0010_KKKK.csv.gz.

我想要一个来自所有文件的单个 CSV 文件，如上所示。请帮助我如何解压缩和合并所有文件。

将所有文件解压并合并为一个 CSV 文件后，我可以使用此文件与以前的文件进行数据比较。

我正在使用 spark 2.3.0 和 scala 2.11

非常感谢。

【参考方案1】：

下面提到的代码似乎工作正常。

scala> val rdd = sc.textFile("/root/data")
rdd: org.apache.spark.rdd.RDD[String] = /root/data MapPartitionsRDD[1] at textFile at <console>:24


scala> rdd.coalesce(1).saveAsTextFile("/root/combinedCsv", classOf[org.apache.hadoop.io.compress.GzipCodec])

可以看到输入数据在/root/data目录下，合并成gzip格式的csv存储在/root/combinedCsv目录下。

更新

如果您想以 csv 格式存储数据，请去掉 GzipCodec 部分。

scala> rdd.coalesce(1).saveAsTextFile("/root/combinedCsv")

【讨论】：

您好，感谢您的帮助。我想解压缩 csv 文件以进一步比较数据。

可以使用gunzip <file location>解压gz文件。或者，您可以通过运行 rdd.coalesce(1).saveAsTextFile("/root/combinedCsv")

以 csv 格式保存 RDD

【参考方案2】：

您可以使用下面的代码，也可以直接从gz文件中读取而不提取：

val filePath = "/home/harneet/<Dir where all gz/csv files are present>"

var cdnImpSchema = StructType(Array(
 StructField("idate",     TimestampType, true),
 StructField("time",     StringType, true),
 StructField("anyOtherColumn",  StringType, true)
))

var cdnImpDF = spark.read.format("csv").     // Use "csv" regardless of TSV or CSV.
 option("delimiter", ","). // Set delimiter to tab or comma or whatever you want.
 schema(cdnImpSchema).        // Schema that was built above.
 load(filePath)

cdnImpDF.repartition(1).write.format("com.databricks.spark.csv").option("header", "true").save("mydata.csv")

repartition(1) -> 将生成一个文件作为输出。