проверка CSV по определению JSON

Question

Я новичок в Scala и задаюсь вопросом, какие будут лучшие методы для проверки файла CSV, предпочтительно с использованием функции карты и добавления нового столбца в зависимости от того, были ли выполнены условия. Я хочу поместить это как функцию UDF для моего фрейма данных в Apache Spark.

Вот схема:

  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"]
     }
  ]
}

Answer 1

Я не уверен в вашем определении JSON (в нем также отсутствуют некоторые кавычки и фигурные скобки) и в том, является ли тип записи столбцом в CSV, но вот упрощение - вы можете добавить логи "Тип записи" c вокруг него, если необходимо.

Предполагается, что файл валидатор. 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|
+----+----+----+-------+