根据 JSON 定义验证 csv答案

【问题标题】：validation of a csv against JSON definition根据 JSON 定义验证 csv
【发布时间】：2020-03-02 09:38:25
【问题描述】：

我是 Scala 新手，想知道验证 CSV 文件的最佳方法是什么最好使用地图功能并根据条件是否满足添加新列。我想把它作为我的数据框在 Apache Spark 中的 UDF 函数。

这是架构：

  Record Type  val1  val2  val3  

   TYPE1         1    2    ZZ 
   TYPE2         2   555   KK

我想要验证的 JSON 定义：

"rows" :
{ 
  "TYPE1" :

   "fields" : [
    {
       "required" : "true",
       "regex":  "TYPE1",
    },
     {
        "required" : true",
         "regex" :"[a-zA-Z]{2}[a-zA-Z]{2}",
       "allowed_values": null  

     },
       {
        "required" : true",
         "regex" :"[a-zA-Z]{2}[a-zA-Z]{2}",
          "allowed_values" : ["ZZ","KK"]
     }
  ]
}

【问题讨论】：

标签： scala

【解决方案1】：

我不确定您的 JSON 定义（它还缺少一些引号和花括号），以及记录类型是否是 CSV 中的列，但这里有一个简化——您可以在它周围添加“记录类型”逻辑如果需要。

假设一个文件validator.json：

{
  "fields" : [
    {
      "name" : "val1",
      "regex": "[0-9]+"
    },{
      "name" : "val2",
      "regex" :"[0-9]+"
    },{
      "name" : "val3",
      "regex" :"[A-Z]{2}"
    }
  ]
}

通常，默认情况下（没有关于架构的额外选项）spark.read.format("csv").option("header", "true").load("file.csv") 将为文件中的所有列使用字符串。在这里，假设您有一个标题val1,val2,val3，作为 CSV 的第一行。一个等效定义的 DF 内联：

val df = Seq(("1", "2", "ZZ"), ("2", "555", "KK")).toDF("val1", "val2", "val3")

import com.fasterxml.jackson.module.scala.DefaultScalaModule
import com.fasterxml.jackson.databind.ObjectMapper
import scala.io.Source

val mapper = new ObjectMapper
mapper.registerModule(DefaultScalaModule)
// read the validator as one long string
val jsonString = Source.fromFile("validator.json").getLines.mkString("")

// map the json string into an object (nested map)
val regexMap:Map[String,Seq[Map[String,String]]] = mapper.readValue(jsonString, classOf[Map[String, Seq[Map[String, String]]]])

//val1 rlike '[0-9]+' AND val2 rlike '[0-9]+' AND val3 rlike '[A-Z]{2}'
val exprStr:String = regexMap("fields").map((fieldDef:Map[String, String]) => s"${fieldDef("name")} rlike '${fieldDef("regex")}'").mkString(" AND ")

// this asks whether all rows match
val matchingRowCount:Long = df.filter(expr("val1 rlike '[0-9]+' AND val2 rlike '[0-9]+' AND val3 rlike '[A-Z][A-Z]'")).count

// if the counts match, then all of the rows follow the rules
df.count == matchingRowCount

// this adds a column about whether the row matches
df.withColumn("matches",expr(exprStr)).show

结果：

+----+----+----+-------+
|val1|val2|val3|matches|
+----+----+----+-------+
|   1|   2|  ZZ|   true|
|   2| 555|  KK|   true|
+----+----+----+-------+

【讨论】：