Spark 使用 S3 的自定义路径写入压缩的 CSV

Question

我正在尝试使用用 Scala 编写的 Spark 将 CSV 简单地写入 S3：

我注意到我的输出存储桶中有以下文件： ...PROCESSED/montfh-04.csv/part-00000-723a3d72-56f6-4e62-b627-9a181a820f6a-c000.csv.snappy

什么时候应该只是montfh-04.csv

代码：

    val processedMetadataDf = spark.read.csv("s3://" + metadataPath + "/PROCESSED/" + "month-04" + ".csv")
    val processCount = processedMetadataDf.count()
    if (processCount == 0) {
        // Initial frame is 0B -> Overwrite with path 
        val newDat = Seq("dummy-row-data")
        val unknown_df = newDat.toDF()
        unknown_df.write.mode("overwrite").option("header","false").csv("s3://" + metadataPath + "/PROCESSED/" + "montfh-04" + ".csv")
        
    }

这里我注意到两件奇怪的事情：

• 它把它放在一个目录中
• 它通过快速压缩将奇怪的部分子路径添加到文件中

我要做的只是将具有该名称的平面 CSV 文件写入指定路径。我有哪些选择？

Answer 1

这就是火花的工作原理。您为保存 DataSet/DataFrame 提供的位置是 spark 可以写入其所有分区的目录位置。 部分文件的数量将等于分区的数量，在您的情况下仅为 1。

现在，如果您只希望文件名是 montfh-04.csv，那么您可以重命名它。

注意：在 S3 中重命名是昂贵的操作（复制和删除）。当您使用 spark 编写时，它将是 I/O 的 3 倍，因为 2 倍将是输出提交操作和 1 倍重命名。最好将其写入 HDFS 并使用所需的密钥名称从那里上传。