scala - Spark SQL - 使用模式读取 csv

Question

我在尝试使用 Spark 来简单读取 CSV 文件时遇到了这个问题。在这样的操作之后，我想确保:

数据类型正确(使用提供的架构)

header 针对提供的架构是正确的

这是我使用的代码并且有以下问题:

val schema = Encoders.product[T].schema
val df = spark.read
 .schema(schema)
 .option("header", "true")
 .csv(fileName)

型号 T类型为 Product ， 一世。 e.案例类。这有效，但它不检查是否 列名是正确的，所以我可以提供另一个文件，只要数据类型正确，错误就不会发生，我不知道用户提供了错误的文件，但与正确的数据类型和正确的排序相吻合。

我尝试使用推断模式的选项，然后使用 .as[T]数据集上的方法，但在除 String 之外的任何列的情况下只包含 null 它被 Spark 解释为 String列，但在我的架构中是 Integer .所以发生了强制转换异常，但列名已经被检查好了。

总结一下:我找到了可以确保正确的数据类型但没有标题的解决方案，以及其他可以验证标题但在数据类型方面存在问题的解决方案。有什么办法可以同时实现，我。 e.标题和类型的完整验证？

我正在使用 Spark 2.2.0。

Answer 1

看起来你必须自己通过阅读文件头两次来完成。

查看Spark的代码，推断的header完全是忽略 (从来没有真正阅读过)如果用户提供了他们自己的模式，所以没有办法让 Spark 在这种不一致的情况下失败。

要自己执行此比较:

val schema = Encoders.product[T].schema

// read the actual schema; This shouldn't be too expensive as Spark's
// laziness would avoid actually reading the entire file 
val fileSchema = spark.read
  .option("header", "true")
  .csv("test.csv").schema

// read the file using your own schema. You can later use this DF
val df = spark.read.schema(schema)
  .option("header", "true")
  .csv("test.csv")

// compare actual and expected column names:
val badColumnNames = fileSchema.fields.map(_.name)
  .zip(schema.fields.map(_.name))
  .filter { case (actual, expected) => actual != expected }

// fail if any inconsistency found:
assert(badColumnNames.isEmpty, 
  s"file schema does not match expected; Bad column names: ${badColumnNames.mkString("; ")}")