scala – Spark SQL – 用模式读取csv

发布时间：2020-12-16 18:34:52 所属栏目：安全来源：网络整理

导读：我在尝试使用Spark简单读取CSV文件时遇到了问题.经过这样的操作后,我想确保：数据类型正确(使用提供的架构) 标题对提供的架构是正确的这是我使用的代码,但有以下问题： val schema = Encoders.product[T].schemaval df = spark.read .schema(schema) .opti

我在尝试使用Spark简单读取CSV文件时遇到了问题.经过这样的操作后,我想确保：

>数据类型正确(使用提供的架构)
>标题对提供的架构是正确的

这是我使用的代码,但有以下问题：

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

T型是Product,i.即案例类.这可行,但它不检查列名是否正确,所以我可以给另一个文件,只要数据类型是正确的,不会发生错误,我不知道用户提供了错误的文件,但有些巧合正确的数据类型和正确的顺序.

我尝试使用推断模式然后在数据集上使用.as [T]方法的选项,但是如果除String之外的任何列只包含null,则由Spark解释为String列,但在我的模式中它是Integer.因此会发生强制转换异常,但已经检查了列名称.

总结一下：我找到了解决方案,我可以确保正确的数据类型,但没有标题和其他解决方案,我可以验证标题,但有数据类型的问题.有没有办法实现两者,我.即标题和类型的完整验证？

我正在使用Spark 2.2.0.

解决方法

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

查看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("; ")}")

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!