Spark Dataset加载多个CSV文件，文件头位于文件夹中，如果所有文件中的文件头不相同，则报告不匹配

Question

我正在尝试使用spark 2.1.0 API将多个csv文件从hdfs目录加载到spark DataSet中：

val csvData = spark.read.option("header", "true").csv("csvdatatest/")

在“CSVDataTest”文件夹中有多个csv文件。Spark只从第一个文件中提取header，并将其生成为数据集的模式，忽略其余csv文件的header。例如

hadoop fs -ls /user/kumara91/csvdatatest
Found 2 items
/user/kumara91/csvdatatest/controlfile-2017-10-19.csv
/user/kumara91/csvdatatest/controlfile-2017-10-23.csv

hadoop fs -cat /user/kumara91/csvdatatest/controlfile-2017-10-19.csv
Delivery ID,BroadLog ID,Channel,Address,Event type,Event date,File name

hadoop fs -cat /user/kumara91/csvdatatest/controlfile-2017-10-23.csv
Delivery ID,BroadLog ID,Channel,Address,Event type,Event date,File name,dummycolumn

scala> val csvData = spark.read.option("header", "true").csv("csvdatatest/")
csvData: org.apache.spark.sql.DataFrame = [Delivery ID: string, BroadLog ID: string ... 5 more fields]

scala> csvData.schema.fieldNames
res1: Array[String] = Array(Delivery ID, BroadLog ID, Channel, Address, Event type, Event date, File name)

这里，它只从文件“controlfile-2017-10-19.csv”加载了头，并忽略了其他csv文件中带有额外列“dummycolumn”的头。

但是，我想知道是否有任何方法可以使用Spark API而不遍历每个文件，以及为什么Spark从一个文件读取头而忽略其余的文件。

Answer 1

如果不以某种方式迭代文件，就无法正确读取数据。在大数据中，基于文件的数据源是基于目录的，而CSV的假设是一个目录中的所有文件都具有相同的模式。对于JSON源代码，不存在与.read.option（“mergeschema”，true）等效的代码。

如果只想检查标题，则需要一次处理一个文件。在获得所有文件的列表后，使用所需的任何方法，最简单的方法是使用如下代码获取文件头：

val paths: Seq[String] = ...
val pathsAndHeaders: Seq[(String, String)] = paths.map { path =>
  val header = sc.textFile(path).take(1).collect.head
  (path, header)
}

如果您有许多CSV，一个更有效的版本是在集群中分发路径，但您必须自己读取文件：

val paths: Seq[String] = ...
val pathsAndHeaders: Seq[(String, String)] = sc.parallelize(paths)
  .map { path =>
    val header = // read first line of file at path
    (path, header)
  }
  .collect

现在已经有了路径和头部，可以做任何需要的事情。例如，一旦将文件分组为具有相同头的组，就可以将一系列路径传递给load()以在一次操作中读取它们。