在 spark 中验证 csv 数据的数据类型的正确方法是什么？

Question

我们有一个 JSON 文件作为 spark 程序的输入（它描述了我们要在每一列上检查的架构定义和约束），我想执行一些数据质量检查，例如（非空，唯一）和数据类型验证以及（想检查csv文件是否包含根据json模式的数据？）。

JSON 文件：

{

"id":"1",

"姓名":"员工",

“来源”：“本地”，

"file_type":"文本",

"sub_file_type":"csv",

"分隔符":",",

"路径":"/user/all/dqdata/data/emp.txt",

“列”：[

{"column_name":"empid","datatype":"integer","constraints":["not null","unique"],"values_permitted":["1","2"]},

{"column_name":"empname","datatype":"string","constraints":["not null","unique"],"values_permitted":["1","2"]},

{"column_name":"salary","datatype":"double","constraints":["not null","unique"],"values_permitted":["1","2"]},

{"column_name":"doj","datatype":"date","constraints":["not null","unique"],"values_permitted":["1","2"]},

{"column_name":"location","string":"number","constraints":["not null","unique"],"values_permitted":["1","2"]}

]

}

CSV 输入示例：

empId,empname,salar,dob,location

1,a,10000,11-03-2019,浦那

2,b,10020,14-03-2019,浦那

3,a,10010,15-03-2019,浦那

a,1,10010,15-03-2019,浦那

请记住，

1) 我故意将无效数据放入 empId 和 name 字段（检查最后一条记录）。 2）json文件的列数不固定？

问题：

如何确保输入数据文件包含给定数据类型（JSON 格式）文件的所有记录？

我们尝试了以下方法：

1) 如果我们尝试通过应用外部模式使用数据框从 CSV 文件中加载数据，那么 spark 程序会立即抛出一些强制转换异常（NumberFormatException 等）并异常终止程序。但我想继续执行流程并将特定错误记录为“列 empID 的数据类型不匹配错误”。 只有当我们在数据帧上调用一些 RDD 操作时，上述场景才有效，我觉得这是一种奇怪的验证模式的方法。

请指导我，我们如何在 spark 中实现它？

Answer 1

我认为这里没有免费的午餐，你必须自己编写这个过程，但你可以做的过程是......

1. 将 csv 文件读取为 Dataset 的 Strings，以便每一行都正确
2. 使用map 函数解析数据集以检查Null 或每列的数据类型问题
3. 添加额外的两列，boolean 称为 validRow 和 String 称为 message 或 description
4. 使用“2.”中提到的解析器，对每列中的每个值执行某种try/catch 或Try/Success/Failure 并捕获异常并相应地设置validRow 和description 列李>
5. 做一个过滤，写一个成功的DataFrame/DataSet（validRow标志设置为True）到成功的地方，把错误的DataFrame/DataSet写到错误的地方