數據質量 — 使用amazon deequ作為spark etl數據質量檢測

目前，公司里數據質量檢測是通過配置規則報警來實現的，對於有些表需要用shell腳本來封裝hivesql來進行檢測，在時效性和准確上不能很好的滿足，故嘗試使用Deequ來做質量檢測工具。

一、官網示例

package org.shydow.deequ

import com.amazon.deequ.checks.CheckStatus
import com.amazon.deequ.constraints.ConstraintStatus
import com.amazon.deequ.{VerificationResult, VerificationSuite}
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * @author shydow
 * @date 2022-03-25
 */


object DQService {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder()
      .appName("DQC")
      .master("local[*]")
      .getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("WARN")
    import spark.implicits._

    val source: RDD[Item] = sc.parallelize(Seq(
      Item(1, "Thingy A", "awesome thing.", "high", 0),
      Item(2, "Thingy B", "available at http://thingb.com", null, 0),
      Item(3, null, null, "low", 5),
      Item(4, "Thingy D", "checkout https://thingd.ca", "low", 10),
      Item(5, "Thingy E", null, "high", 12)))
    val sourceDF: DataFrame = spark.createDataFrame(source)
    sourceDF.printSchema()

    // 質量檢測
    val result: VerificationResult = DeequCheckRules.createRule(sourceDF)
    if (result.status == CheckStatus.Success) {
      println("The data passed the test, everything is fine!")
    } else {
      println("We found errors in the data:\n")

      val resultsForAllConstraints = result.checkResults
        .flatMap { case (_, checkResult) => checkResult.constraintResults }

      resultsForAllConstraints
        .filter {
          _.status != ConstraintStatus.Success
        }
        .foreach { result => println(s"${result.constraint}: ${result.message.get}") }
    }

    spark.close()
  }
}

package org.shydow.deequ

import com.amazon.deequ.{VerificationResult, VerificationSuite}
import com.amazon.deequ.checks.{Check, CheckLevel}
import org.apache.spark.sql.DataFrame

/**
 * @author shydow
 * @date 2022-03-25
 */

object DeequCheckRules {
  // 自定義規則1
  def createRule(df: DataFrame): VerificationResult = {
    VerificationSuite().onData(df)
      .addCheck(Check(CheckLevel.Error, "this a unit test")
        .hasSize(_ == 5) // 判斷數據量是否是5條
        .isComplete("id") // 判斷該列是否全部不為空
        .isUnique("id") // 判斷該字段是否是唯一
        .isComplete("productName") // 判斷該字段全部不為空
        .isContainedIn("priority", Array("high", "low")) // 該字段僅僅包含這兩個字段
        .isNonNegative("numViews") //該字段不包含負數
        .containsURL("description", _ >= 0.5) // 包含url的記錄是否超過0.5
        .hasApproxQuantile("numViews", 0.5, _ <= 10)
      )
      .run()
  }
}

 

二、生產中配置的一些規則

def odsTableRule(df: DataFrame) = {
    VerificationSuite()
      .onData(df)
      .addCheck(
        Check(CheckLevel.Error, "base checks")
          .isComplete("primaryKey") // primaryKey即主要字段不能為空
          .isUnique("uniqueKey") // unique即唯一主鍵
          .isContainedIn("priority", Array("high", "low")) // 判斷該字段是否只存在枚舉類型
          .isNonNegative("numViews") // 斷言該字段非負數
          .satisfies(
            "abs(column1 - column2) <= 0.20 * column2",
            "value(column1) lies between value(column2)-20% and value(column2)+20%"
          )  // 自定義條件，判斷col1-col2絕對值在0.2 * col2間
      )
      .addCheck(
        Check(CheckLevel.Warning, "distribution checks")
          .containsURL("description", _ >= 0.5)  // 斷言有一半的值包含url
          .hasApproxQuantile("numViews", 0.5, _ <= 10))  // 斷言有一半的值不超過10
      .run()
  }