可以将长度与标头记录不同的记录放入 bad_record 目录答案

【问题标题】：Possible to put records that aren't same length as header records to bad_record directory可以将长度与标头记录不同的记录放入 bad_record 目录
【发布时间】：2019-02-02 05:57:18
【问题描述】：

我正在将文件读入这样的数据框

val df = spark.read
   .option("sep", props.inputSeperator)
   .option("header", "true")
   .option("badRecordsPath", "/mnt/adls/udf_databricks/error")
   .csv(inputLoc)

文件是这样设置的

col_a|col_b|col_c|col_d
1|first|last|
2|this|is|data
3|ok
4|more||stuff
5|||

现在，spark 会将所有这些内容作为可接受的数据读取。但是，我希望将 3|ok 标记为错误记录，因为它的大小与标题大小不匹配。这可能吗？

【问题讨论】：

标签： scala apache-spark error-handling apache-spark-sql databricks

【解决方案1】：

val a = spark.sparkContext.textFile(pathOfYourFile)
val size = a.first.split("\\|").length
a.filter(i => i.split("\\|",-1).size != size).saveAsTextFile("/mnt/adls/udf_databricks/error")

【讨论】：

谢谢.. 但是，我提供的代码有效，它会生成一个 BadRecord 输出，只是不包含小于所需大小的行。所以，我会假设这意味着 badRecordsPath 选项存在？（它会在/mnt/adls/udf_databricks/error 内生成一个带时间戳的文件夹以及输出的部分文件。无论哪种方式，您的解决方案都足够了，谢谢。
您使用的是哪个版本的 spark ？让我检查源代码。
版本 2.3.1 //
感谢您带来这个。我不知道那个选项。对此点赞。
没问题，我假设即使csv存在这个选项，你也不能修改它来完成我想要的？

【解决方案2】：

spark 的 databricks 实现支持以下代码。我在您的代码中没有看到架构映射。你能映射它并尝试吗？

.option("badRecordsPath", "/mnt/adls/udf_databricks/error")

如下更改您的代码，

val customSchema = StructType(Array(
    StructField("col_a", StringType, true),
    StructField("col_b", StringType, true),
    StructField("col_c", StringType, true),
    StructField("col_d", StringType, true)))

val df = spark.read
   .option("sep", props.inputSeperator)
   .option("header", "true")
   .option("badRecordsPath", "/mnt/adls/udf_databricks/error")
   .schema(customSchema)
   .csv(inputLoc)

更多细节你可以参考Datbricks doc on badrecordspath

谢谢，卡西克

【讨论】：

好的，这解决了我的问题，但是，必须维护架构会增加一些开销。架构始终是所有字符串类型，基于标题列名称。我不想使用推断模式，因为我有数亿条记录，这将是昂贵的。有什么解决办法吗？
只是补充一下，我不知道列名是什么。我所知道的是我在inputLoc 中有一个标题行，它们都是字符串类型。我宁愿不进行 2 次读取操作，一次获取标题行，然后再次将其读入数据帧。 @Karthick
@testacc：您可以使用普通的 scala 读取标头并动态创建要使用的架构，如上所示。如果它是单个文件，它不会增加那么多开销。