Spark SQL-使用架构读取csv

Question

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

• 数据类型是正确的（使用提供的模式）
• 根据提供的架构，标头是正确的

这是我使用的代码，并且有问题：

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

类型为产品类型，即案例类。这是可行的，但它不会检查列名是否正确，因此我可以提供另一个文件，只要数据类型正确，就不会发生错误，而且我不知道用户提供了错误的文件，但由于某种程度上的巧合，正确的数据类型具有正确的顺序。

我尝试使用推断架构的选项，然后在Dataset上使用. as[T]方法，但如果String以外的任何列仅包含null，则Spark将其解释为String列，但在我的架构中它是整数。所以发生了强制转换异常，但列名已检查无误。

总结一下：我找到了可以确保正确数据类型但没有标头的解决方案，以及可以验证标头但数据类型有问题的其他解决方案。有没有办法同时实现这两个目标，即。e.完全验证标头和类型？

我正在使用Spark 2.2.0。

Answer 1

看起来您必须自己读取两次文件头。

看看Spark的代码，如果用户提供他们自己的模式，那么推断出的头将被完全忽略（从未真正读取），因此无法使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("; ")}")