使用 Delta Live Tables 管理数据质量

项目
03/21/2024

使用预期来定义对数据集内容的数据质量约束。期望使你能够保证进入表中的数据满足数据质量要求，并为每项管道更新提供数据质量的见解。使用 Python 修饰器或 SQL 约束子句将预期应用于查询。

什么是增量实时表期望？

期望是添加到增量实时表数据集声明的可选子句，这些子句对通过查询的每条记录应用数据质量检查。

期望由三个元素组成：

一段说明，它充当唯一标识符，并可用于跟踪约束的指标。
一个布尔语句，它始终根据某个规定的条件返回 true 或 false。
当记录不满足期望（即布尔语句返回 false）时要执行的操作。

以下矩阵显示了可应用于无效记录的三个操作：

操作	结果
警告（默认值）	将无效记录写入目标；将失败报告为数据集的指标。
删除	在将数据写入目标之前删除无效记录；将失败报告为数据集的指标。
失败	无效记录会阻止更新成功完成。在重新处理之前需要手动干预。

你可以通过查询增量实时表事件日志来查看数据质量指标，例如违反预期的记录数。请参阅监视增量实时表管道。

有关增量实时表数据集声明语法的完整参考，请参阅增量实时表 Python 语言参考或增量实时表 SQL 语言参考。

注意

虽然可以在任一期望中包含多个子句，但只有 Python 支持基于多个期望定义操作。请参阅多个期望。

保留无效记录

当你想要保留不符合预期的记录时，请使用 expect 运算符。不符合预期的记录将与有效记录一起添加到目标数据集中：

Python

@dlt.expect("valid timestamp", "col(“timestamp”) > '2012-01-01'")

SQL

CONSTRAINT valid_timestamp EXPECT (timestamp > '2012-01-01')

删除无效记录

使用 expect or drop 运算符可防止进一步处理无效记录。不符合预期的记录将从目标数据集中删除：

Python

@dlt.expect_or_drop("valid_current_page", "current_page_id IS NOT NULL AND current_page_title IS NOT NULL")

SQL

CONSTRAINT valid_current_page EXPECT (current_page_id IS NOT NULL and current_page_title IS NOT NULL) ON VIOLATION DROP ROW

出现无效记录时失败

当无效记录不可接受时，使用 expect or fail 运算符在记录未通过验证时立即停止执行。如果这项操作是表更新，系统会自动回滚事务：

Python

@dlt.expect_or_fail("valid_count", "count > 0")

SQL

CONSTRAINT valid_count EXPECT (count > 0) ON VIOLATION FAIL UPDATE

当管道因预期违规而失败时，必须在重新运行管道之前修复管道代码，以正确处理无效数据。

预期违规会修改转换的 Spark 查询计划，以跟踪检测和报告违规所需的信息。对于许多查询，可以使用此信息来识别导致违规的输入记录。下面是一个异常示例：

Expectation Violated:
{
  "flowName": "a-b",
  "verboseInfo": {
    "expectationsViolated": [
      "x1 is negative"
    ],
    "inputData": {
      "a": {"x1": 1,"y1": "a },
      "b": {
        "x2": 1,
        "y2": "aa"
      }
    },
    "outputRecord": {
      "x1": 1,
      "y1": "a",
      "x2": 1,
      "y2": "aa"
    },
    "missingInputData": false
  }
}

多个预期

可以在 Python 管道中定义具有一个或多个数据质量约束的期望。这些修饰器接受 Python 字典作为参数，其中键是预期名称，值是预期约束。

当未通过验证的记录应包含在目标数据集中时，请使用 expect_all 指定多个数据质量约束：

@dlt.expect_all({"valid_count": "count > 0", "valid_current_page": "current_page_id IS NOT NULL AND current_page_title IS NOT NULL"})

当未通过验证的记录应从目标数据集中删除时，请使用 expect_all_or_drop 指定多个数据质量约束：

@dlt.expect_all_or_drop({"valid_count": "count > 0", "valid_current_page": "current_page_id IS NOT NULL AND current_page_title IS NOT NULL"})

当未通过验证的记录应停止管道执行时，请使用 expect_all_or_fail 指定多个数据质量约束：

@dlt.expect_all_or_fail({"valid_count": "count > 0", "valid_current_page": "current_page_id IS NOT NULL AND current_page_title IS NOT NULL"})

你还可以将一组预期定义为变量，并将其传递给管道中的一个或多个查询：

valid_pages = {"valid_count": "count > 0", "valid_current_page": "current_page_id IS NOT NULL AND current_page_title IS NOT NULL"}

@dlt.table
@dlt.expect_all(valid_pages)
def raw_data():
  # Create raw dataset

@dlt.table
@dlt.expect_all_or_drop(valid_pages)
def prepared_data():
  # Create cleaned and prepared dataset

隔离无效数据

以下示例将期望与临时表和视图结合使用。此模式提供在管道更新期间通过期望检查的记录的指标，并提供一种通过不同下游路径处理有效和无效记录的方式。

注意

此示例读取 Databricks 数据集中包含的示例数据。由于发布到 Unity Catalog 的管道不支持 Databricks 数据集，因此此示例仅适用于配置为发布到 Hive 元存储的管道。不过，此模式也适用于已启用 Unity Catalog 的管道，但你必须从外部位置读取数据。若要详细了解如何将 Unity Catalog 与 Delta Live Tables 配合使用，请参阅将 Unity Catalog 与 Delta Live Tables 管道配合使用。

import dlt
from pyspark.sql.functions import expr

rules = {}
rules["valid_website"] = "(Website IS NOT NULL)"
rules["valid_location"] = "(Location IS NOT NULL)"
quarantine_rules = "NOT({0})".format(" AND ".join(rules.values()))

@dlt.table(
  name="raw_farmers_market"
)
def get_farmers_market_data():
  return (
    spark.read.format('csv').option("header", "true")
      .load('/databricks-datasets/data.gov/farmers_markets_geographic_data/data-001/')
  )

@dlt.table(
  name="farmers_market_quarantine",
  temporary=True,
  partition_cols=["is_quarantined"]
)
@dlt.expect_all(rules)
def farmers_market_quarantine():
  return (
    dlt.read("raw_farmers_market")
      .select("MarketName", "Website", "Location", "State",
              "Facebook", "Twitter", "Youtube", "Organic", "updateTime")
      .withColumn("is_quarantined", expr(quarantine_rules))
  )

@dlt.view(
  name="valid_farmers_market"
)
def get_valid_farmers_market():
  return (
    dlt.read("farmers_market_quarantine")
      .filter("is_quarantined=false")
  )

@dlt.view(
  name="invalid_farmers_market"
)
def get_invalid_farmers_market():
  return (
    dlt.read("farmers_market_quarantine")
      .filter("is_quarantined=true")
  )

验证跨表的行计数

可以向管道添加一个附加表，用于定义期望来比较两个实时表之间的行计数。该预期的结果显示在事件日志和增量实时表 UI 中。以下示例验证 tbla 和 tblb 表之间的行计数是否相等：

CREATE OR REFRESH LIVE TABLE count_verification(
  CONSTRAINT no_rows_dropped EXPECT (a_count == b_count)
) AS SELECT * FROM
  (SELECT COUNT(*) AS a_count FROM LIVE.tbla),
  (SELECT COUNT(*) AS b_count FROM LIVE.tblb)

根据 Delta Live Tables 预期执行高级验证

可使用聚合和联接查询定义 Live Tables，并将这些查询的结果用作期望检查的一部分。如果你想要执行复杂的数据质量检查，从而确保派生表包含源表中的所有记录或保证表中数值列的相等性，则此功能非常有用。可以使用 TEMPORARY 关键字来防止将这些表发布到目标架构。

以下示例验证 report 表中是否存在所有预期记录：

CREATE TEMPORARY LIVE TABLE report_compare_tests(
  CONSTRAINT no_missing_records EXPECT (r.key IS NOT NULL)
)
AS SELECT * FROM LIVE.validation_copy v
LEFT OUTER JOIN LIVE.report r ON v.key = r.key

以下示例使用聚合来确保主键的唯一性：

CREATE TEMPORARY LIVE TABLE report_pk_tests(
  CONSTRAINT unique_pk EXPECT (num_entries = 1)
)
AS SELECT pk, count(*) as num_entries
FROM LIVE.report
GROUP BY pk

使期望可移植且可重用

可以将数据质量规则与管道实现分开维护。

Databricks 建议将规则存储在 Delta 表中，每个规则按标记分类。在数据集定义中使用此标记来确定要应用的规则。

以下示例创建一个名为 rules 的表来维护规则：

CREATE OR REPLACE TABLE
  rules
AS SELECT
  col1 AS name,
  col2 AS constraint,
  col3 AS tag
FROM (
  VALUES
  ("website_not_null","Website IS NOT NULL","validity"),
  ("location_not_null","Location IS NOT NULL","validity"),
  ("state_not_null","State IS NOT NULL","validity"),
  ("fresh_data","to_date(updateTime,'M/d/yyyy h:m:s a') > '2010-01-01'","maintained"),
  ("social_media_access","NOT(Facebook IS NULL AND Twitter IS NULL AND Youtube IS NULL)","maintained")
)

以下 Python 示例基于 rules 表中存储的规则定义数据质量期望。 get_rules() 函数从 rules 表读取规则并返回一个 Python 字典，其中包含与传递给该函数的 tag 参数匹配的规则。该字典应用于 @dlt.expect_all_*() 修饰器以强制实施数据质量约束。例如，将从 raw_farmers_market 表中删除任何与使用 validity 标记的规则不符的记录：

注意

import dlt
from pyspark.sql.functions import expr, col

def get_rules(tag):
  """
    loads data quality rules from a table
    :param tag: tag to match
    :return: dictionary of rules that matched the tag
  """
  rules = {}
  df = spark.read.table("rules")
  for row in df.filter(col("tag") == tag).collect():
    rules[row['name']] = row['constraint']
  return rules

@dlt.table(
  name="raw_farmers_market"
)
@dlt.expect_all_or_drop(get_rules('validity'))
def get_farmers_market_data():
  return (
    spark.read.format('csv').option("header", "true")
      .load('/databricks-datasets/data.gov/farmers_markets_geographic_data/data-001/')
  )

@dlt.table(
  name="organic_farmers_market"
)
@dlt.expect_all_or_drop(get_rules('maintained'))
def get_organic_farmers_market():
  return (
    dlt.read("raw_farmers_market")
      .filter(expr("Organic = 'Y'"))
      .select("MarketName", "Website", "State",
        "Facebook", "Twitter", "Youtube", "Organic",
        "updateTime"
      )
  )

可以将 Python 模块创建到主规则中，例如，在笔记本所在的同一文件夹中名为 rules_module.py 的文件中创建 Python 模块，而不是创建名为 rules 的表来维护规则：

def get_rules_as_list_of_dict():
  return [
    {
      "name": "website_not_null",
      "constraint": "Website IS NOT NULL",
      "tag": "validity"
    },
    {
      "name": "location_not_null",
      "constraint": "Location IS NOT NULL",
      "tag": "validity"
    },
    {
      "name": "state_not_null",
      "constraint": "State IS NOT NULL",
      "tag": "validity"
    },
    {
      "name": "fresh_data",
      "constraint": "to_date(updateTime,'M/d/yyyy h:m:s a') > '2010-01-01'",
      "tag": "maintained"
    },
    {
      "name": "social_media_access",
      "constraint": "NOT(Facebook IS NULL AND Twitter IS NULL AND Youtube IS NULL)",
      "tag": "maintained"
    }
  ]

然后，通过导入模块并将 get_rules() 函数更改为从模块中读取，而不是从 rules 表中读取来修改前面的笔记本：

import dlt
from rules_module import *
from pyspark.sql.functions import expr, col

df = spark.createDataFrame(get_rules_as_list_of_dict())

def get_rules(tag):
  """
    loads data quality rules from a table
    :param tag: tag to match
    :return: dictionary of rules that matched the tag
  """
  rules = {}
  for row in df.filter(col("tag") == tag).collect():
    rules[row['name']] = row['constraint']
  return rules

@dlt.table(
  name="raw_farmers_market"
)
@dlt.expect_all_or_drop(get_rules('validity'))
def get_farmers_market_data():
  return (
    spark.read.format('csv').option("header", "true")
      .load('/databricks-datasets/data.gov/farmers_markets_geographic_data/data-001/')
  )

@dlt.table(
  name="organic_farmers_market"
)
@dlt.expect_all_or_drop(get_rules('maintained'))
def get_organic_farmers_market():
  return (
    dlt.read("raw_farmers_market")
      .filter(expr("Organic = 'Y'"))
      .select("MarketName", "Website", "State",
        "Facebook", "Twitter", "Youtube", "Organic",
        "updateTime"
      )
  )

使用 Delta Live Tables 管理数据质量

什么是增量实时表期望？

保留无效记录

Python

SQL

删除无效记录

Python

SQL

出现无效记录时失败

Python

SQL

多个预期

隔离无效数据

验证跨表的行计数

根据 Delta Live Tables 预期执行高级验证

使期望可移植且可重用

其他资源