使用dataframe.schema（val header=datadf.schema.fieldnames.reduce（“+”，“+”）
在dsefs上创建一个头文件
使用hadoop文件系统api将所有分区文件（无头）附加到#2中的文件

你可以这样走来走去。
1.创建包含标头名称的新Dataframe（headerdf）。
2.将其与包含数据的Dataframe（datadf）合并。
3.使用选项（“header”，“false”）将union-edDataframe输出到磁盘。
4.使用hadoop fileutil合并分区文件（part-0000**0.csv）
通过这种方式，除了单个分区的内容之外，所有分区都没有头，头df中有一行头名称。当所有分区合并在一起时，文件的顶部只有一个头。示例代码如下

//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)

尝试指定头的架构，并使用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中，您将只有头的行，从中您可以按照您需要的方式转换Dataframe。

要将文件夹中的文件合并到一个文件中，请执行以下操作：

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
}

我们遇到了一个类似的问题，按照下面的方法获取单个输出文件-
将Dataframe写入带标头的hdfs，而不使用 coalesce 或者 repartition （转换后）。

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

读取上一步中的文件，并使用 coalesce(1) .

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）。第二步提供带有一个标题行的单个输出文件。

// 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/