使用单个标头合并 Spark 输出 CSV 文件答案

【问题标题】：Merge Spark output CSV files with a single header使用单个标头合并 Spark 输出 CSV 文件
【发布时间】：2016-10-29 14:19:43
【问题描述】：

我想在 AWS 中创建一个数据处理管道，最终将处理后的数据用于机器学习。

我有一个 Scala 脚本，它从 S3 获取原始数据，对其进行处理并使用 Spark-CSV 将其写入 HDFS 甚至 S3。如果我想使用 AWS 机器学习 工具来训练预测模型，我想我可以使用多个文件作为输入。但如果我想使用其他东西，我认为最好接收单个 CSV 输出文件。

目前，由于出于性能目的我不想使用 repartition(1) 或 coalesce(1)，因此我使用了 hadoop fs -getmerge 用于手动测试，但由于它只是合并作业输出文件的内容，我遇到了一个小问题。我需要在数据文件中单行标题来训练预测模型。

如果我对 spark-csv 使用 .option("header","true")，那么它将标头写入每个输出文件，合并后，我在数据中拥有与输出文件一样多的标头行。但如果 header 选项为 false，则不会添加任何 headers。

现在我找到了一个选项，可以将 Scala 脚本中的文件与 Hadoop API FileUtil.copyMerge 合并。我用下面的代码在spark-shell 中尝试了这个。

import org.apache.hadoop.fs.FileUtil
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
val configuration = new Configuration();
val fs = FileSystem.get(configuration);
FileUtil.copyMerge(fs, new Path("smallheaders"), fs, new Path("/home/hadoop/smallheaders2"), false, configuration, "")

但是这个解决方案仍然只是将文件连接在一起并且不处理标题。 如何获得只有一行标题的输出文件？

我什至尝试添加 df.columns.mkString(",") 作为 copyMerge 的最后一个参数，但这仍然多次添加标题，而不是一次。

【问题讨论】：

我也面临同样的问题。这个问题解决了吗？
如何将 DataFrame 过滤为零行，使用 header=true 导出，使用 header=false 导出其余数据，然后将 header 与分区合并？
@Boern 这可能有效。虽然我认为这需要将头文件复制到与数据相同的输出并确保它始终是第一个文件。我认为这个当前的解决方案不允许写入相同的路径。当然附加可能会解决这个问题，需要尝试一段时间。
@bleka “如何”是这个问题的重点。可以想象一个触发标志，告诉它只保存带有指定文件part-0000 的标题，或者可能是一种智能串联，它将多个工作人员保存的文件组合在一起，但只保留其中一个工作人员的标题。 copyMerge 看起来它只是组合文件，所以如果文件有标题，标题将出现多次，或者如果文件没有标题，则根本没有标题，正如 V. Samma 在问题中所说。还是copyMerge 在您的回答中有不同的行为？
@belka 虽然这些不是具有不同列的不同数据帧，但它们只是具有相同列的同一数据帧的不同分区

标签： scala csv hadoop apache-spark

【解决方案1】：

你可以这样走动。

1.创建一个包含标头名称的新DataFrame(headerDF)。
2.将其与包含数据的 DataFrame(dataDF) 合并。
3.使用option("header", "false")将union-ed DataFrame输出到磁盘。
4.使用hadoop FileUtil合并分区文件(part-0000**0.csv)

通过这种方式，所有分区都没有标题，除了单个分区的内容具有来自 headerDF 的一行标题名称。当所有分区合并在一起时，文件顶部有一个标题。示例代码如下

  //dataFrame is the data to save on disk
  //cast types of all columns to String
  val dataDF = dataFrame.select(dataFrame.columns.map(c => dataFrame.col(c).cast("string")): _*)

  //create a new data frame containing only header names
  import scala.collection.JavaConverters._
  val headerDF = sparkSession.createDataFrame(List(Row.fromSeq(dataDF.columns.toSeq)).asJava, dataDF.schema)

  //merge header names with data
  headerDF.union(dataDF).write.mode(SaveMode.Overwrite).option("header", "false").csv(outputFolder)

  //use hadoop FileUtil to merge all partition csv files into a single file
  val fs = FileSystem.get(sparkSession.sparkContext.hadoopConfiguration)
  FileUtil.copyMerge(fs, new Path(outputFolder), fs, new Path("/folder/target.csv"), true, spark.sparkContext.hadoopConfiguration, null)

【讨论】：

【解决方案2】：

使用 dataframe.schema 输出标头 ( val header = dataDF.schema.fieldNames.reduce(_ + "," + _))
在 dsefs 上创建带有标题的文件
使用 hadoop Filesystem API 将所有分区文件（无头文件）附加到 #2 中的文件

【讨论】：

【解决方案3】：

我们遇到了类似的问题，按照以下方法获取单个输出文件-

在不使用coalesce 或repartition（转换后）的情况下将数据帧写入hdfs。

dataframe.write.format("csv").option("header", "true").save(hdfs_path_for_multiple_files)

读取上一步中的文件并使用coalesce(1)写回hdfs上的不同位置。

dataframe = spark.read.option('header', 'true').csv(hdfs_path_for_multiple_files)

dataframe.coalesce(1).write.format('csv').option('header', 'true').save(hdfs_path_for_single_file)

这样，您将避免在执行转换（第 1 步）时与合并或重新分区相关的性能问题。第二步提供带有一个标题行的单个输出文件。

【讨论】：

嗨@Arati Nagmal，我目前不再使用Spark，所以我很难对其进行测试。我实际上会对这些转换感到好奇，并希望看到一些测试结果，也许需要多少时间才能将第一种和第二种方法合并并进行比较。因为对我来说，再次写入、读取和写入相同的数据似乎是相同结果的两倍工作。当然，如果您对数据帧应用了一些转换，第一次写入会比第二次花费更多时间，但是写入两次而不是第一次尝试合并会更快吗？
嗨@V.Samma，是的，这种方法在我们的案例中非常高效。将 coalesce(1) 与转换一起使用的问题在于，它以并行度 1 执行转换。并且需要大量时间才能完成。这就是为什么我分为两个步骤。在我们的例子中，第二步大约需要 2 分钟，即 CSV 格式的数据约为 3GB。我希望它能澄清你的问题！

【解决方案4】：

将文件夹中的文件合并为一个文件：

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs._

def merge(srcPath: String, dstPath: String): Unit =  {
  val hadoopConfig = new Configuration()
  val hdfs = FileSystem.get(hadoopConfig)
  FileUtil.copyMerge(hdfs, new Path(srcPath), hdfs, new Path(dstPath), false, hadoopConfig, null)
}

如果您想将所有文件合并到一个文件中，但仍位于同一文件夹中（但这会将所有数据带到驱动程序节点）：

dataFrame
      .coalesce(1)
      .write
      .format("com.databricks.spark.csv")
      .option("header", "true")
      .save(out)

另一种解决方案是使用解决方案 #2，然后将文件夹内的一个文件移动到另一个路径（使用我们的 CSV 文件的名称）。

def df2csv(df: DataFrame, fileName: String, sep: String = ",", header: Boolean = false): Unit = {
    val tmpDir = "tmpDir"

    df.repartition(1)
      .write
      .format("com.databricks.spark.csv")
      .option("header", header.toString)
      .option("delimiter", sep)
      .save(tmpDir)

    val dir = new File(tmpDir)
    val tmpCsvFile = tmpDir + File.separatorChar + "part-00000"
    (new File(tmpCsvFile)).renameTo(new File(fileName))

    dir.listFiles.foreach( f => f.delete )
    dir.delete
}

【讨论】：

【解决方案5】：

尝试使用 spark-csv 的 drop malformed 选项指定标头的架构并从文件夹中读取所有文件。这应该让您读取文件夹中的所有文件，只保留标题（因为您删除了格式错误的文件）。示例：

val headerSchema = List(
  StructField("example1", StringType, true),
  StructField("example2", StringType, true),
  StructField("example3", StringType, true)
)

val header_DF =sqlCtx.read
  .option("delimiter", ",")
  .option("header", "false")
  .option("mode","DROPMALFORMED")
  .option("inferSchema","false")
  .schema(StructType(headerSchema))
  .format("com.databricks.spark.csv")
  .load("folder containg the files")

在 header_DF 中，您将只有标题的行，由此您可以按照您需要的方式转换数据框。

【讨论】：

虽然我不再使用 Spark，但我没有可以测试它的系统，但您的示例可能适用于较小的结构。但是我有一个案例，我们在一个结构中有 200-300 个字段（结构嵌套在其他结构中等），并且随着系统的变化，这个模式变化很快。因此，手动定义架构不是一种选择。

【解决方案6】：

 // Convert JavaRDD  to CSV and save as text file
        outputDataframe.write()
                .format("com.databricks.spark.csv")
                // Header => true, will enable to have header in each file
                .option("header", "true")

请点击集成测试链接了解如何编写单个标头

http://bytepadding.com/big-data/spark/write-a-csv-text-file-from-spark/

【讨论】：