DataHub Snowflake DMF Assertions 实战指南：用 YAML 声明式断言编译为 Snowflake Data Metric Functions

DataHub Snowflake DMF Assertions 实战指南：用 YAML 声明式断言编译为 Snowflake Data Metric Functions

【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub

本指南围绕 DataHub 的 Open Assertion Compiler 展开，讲解如何用简单的 YAML 文件声明数据质量断言，并将其编译为可在 Snowflake 上原生执行的 Data Metric Functions（DMF），再通过 Snowflake 摄取把执行结果回传到 DataHub，在数据集上下文中以历史时间线的形式呈现。读完本文，你将掌握从 YAML 定义、CLI 注册、DMF 编译、Snowflake 侧注册执行，到摄取断言结果的完整闭环流程，并理解外部（用户自建）DMF 的接入方式与约束边界。

该功能当前处于BETA状态。

一、整体思路：从 YAML 到 Snowflake DMF

DataHub 的 Open Assertion Compiler 允许你：

1. 以简单的 YAML 格式声明数据质量断言；
2. 将断言编译成 Snowflake Data Metric Functions（DMF）；
3. 在 Snowflake 环境中注册编译产物；
4. 在常规 DataHub 摄取流程中把 DMF 执行结果拉回 DataHub，作为普通 Assertion Results 展示在关联表的历史时间线上。

编译与执行是分离的：compile命令只负责把 YAML 断言翻译成 SQL 工件，不连接 Snowflake、不执行任何查询；注册与调度由 Snowflake 侧完成；结果回传则发生在 DataHub 的 Snowflake ingestion 过程中。

从源码看，Snowflake 编译器的实现位于 compiler.py，其内部由三部分组成：SnowflakeMetricSQLGenerator（生成指标 SQL）、SnowflakeMetricEvalOperatorSQLGenerator（生成评估条件 SQL）、SnowflakeDMFHandler（拼装CREATE DATA METRIC FUNCTION与ALTER TABLE ... ADD DATA METRIC FUNCTION语句）。

二、前置条件

开始之前，请确认以下条件全部满足：

• 拥有Snowflake Enterprise 账户，且已启用 DMF 功能；
• 具备在 Snowflake 环境中预置（provision）DMF的权限（见下文权限章节）；
• 具备在 Snowflake 环境中查询 DMF 结果的权限（见下文权限章节）；
• 已有一个完成 Snowflake 元数据摄取、可用的 DataHub 实例。如果尚未配置 Snowflake 摄取，参考 Snowflake Quickstart Guide 上手；
• 已安装 DataHub CLI 并运行过datahub init。

三、权限准备

在 Snowflake 侧正确配置权限是整套流程成功的前提。以下是三组不同角色所需的最小权限集合。

3.1 注册 DMF 所需的权限

执行 DMF 注册与取摄取的服务账户（service account）必须具备：

同时必须授予的角色：

3.2 运行 DMF（调度执行）所需的权限

由于定时调度的 DMF 以表属主（table owner）的角色运行，表属主必须具备：

同时必须授予的角色：

3.3 查询 DMF 结果所需的权限

此外，执行 DataHub Ingestion 并查询 DMF 结果的服务账户，还必须被授予以下系统应用角色：

有关 Snowflake DMF 及其预置、查询所需权限的更多细节，参见 Snowflake 官方数据质量文档。

3.4 授权示例 SQL

以下 SQL 给出了一个可直接套用的授权脚本，覆盖上述三类角色（<assertion-service-role>、<table-owner-role>、<datahub_role>）：

-- 为 <assertion-service-role> 配置创建 DMF 并与表关联的权限 grant usage on database "<dmf-database>" to role "<assertion-service-role>" grant usage on schema "<dmf-database>.<dmf-schema>" to role "<assertion-service-role>" grant create function on schema "<dmf-database>.<dmf-schema>" to role "<assertion-service-role>" -- 授予 <assertion-service-role> 所有权及其余权限 grant role "<table-owner-role>" to role "<assertion-service-role>" -- 为 <table-owner-role> 配置按计划运行 DMF 的权限 grant usage on database "<dmf-database>" to role "<table-owner-role>" grant usage on schema "<dmf-database>.<dmf-schema>" to role "<table-owner-role>" grant usage on all functions in "<dmf-database>.<dmf-schema>" to role "<table-owner-role>" grant usage on future functions in "<dmf-database>.<dmf-schema>" to role "<table-owner-role>" grant database role SNOWFLAKE.DATA_METRIC_USER to role "<table-owner-role>" grant execute data metric function on account to role "<table-owner-role>" -- 为 <datahub-role> 配置查询 DMF 结果的权限 grant application role SNOWFLAKE.DATA_QUALITY_MONITORING_VIEWER to role "<datahub_role>"

注意其中使用了USAGE ON FUTURE FUNCTIONS，确保后续新建的 DMF 也自动对表属主可见。

四、支持的断言类型

DataHub Snowflake DMF Assertion Compiler 当前支持以下断言类型：

• Freshness：校验表是否在指定时间窗口内被更新；
• Volume：校验表行数等体量指标是否满足阈值；
• Column：校验列上的指标（如空值数）或列值约束；
• Custom SQL：自定义 SQL 返回的指标是否满足条件。

注意：Schema Assertions 目前不受支持。

从源码看，编译器对断言类型的支持体现在 metric_sql_generator.py 的singledispatchmethod分发中：RowCountChangeVolumeAssertion与SqlMetricChangeAssertion（增量/变更型断言）会被显式判定为Unsupported assertion type并抛错，这与文档中"变更类断言不被支持"的限制一致。

五、完整的五步实战流程

整个流程分为五步：定义 YAML 断言 → 注册到 DataHub → 编译为 Snowflake DMF → 在 Snowflake 注册执行 → 摄取结果回传。

Step 1：用 Assertion YAML 文件定义数据质量断言

断言以 YAML 声明，仓库提供了一个完整的可运行示例 assertions_configuration.yml，包含五种典型断言：

version: 1 namespace: test-config-id-1 assertions: # Freshness Assertion - entity: urn:li:dataset:(urn:li:dataPlatform:snowflake,test_db.public.test_assertions_all_times,PROD) type: freshness lookback_interval: "1 hour" last_modified_field: col_timestamp schedule: type: cron cron: 0 * * * * meta: entity_qualified_name: TEST_DB.PUBLIC.TEST_ASSERTIONS_ALL_TIMES entity_schema: - col: col_date native_type: DATE # Volume Assertion - type: volume entity: urn:li:dataset:(urn:li:dataPlatform:snowflake,test_db.public.test_assertions_all_times,PROD) metric: row_count condition: type: less_than_or_equal_to value: 1000 schedule: type: cron cron: 0 * * * * meta: entity_qualified_name: TEST_DB.PUBLIC.TEST_ASSERTIONS_ALL_TIMES entity_schema: - col: col_date native_type: DATE # Field Metric Assertion - type: field entity: urn:li:dataset:(urn:li:dataPlatform:snowflake,test_db.public.test_assertions_all_times,PROD) field: col_date metric: null_count condition: type: equal_to value: 0 schedule: type: cron cron: 0 * * * * meta: entity_qualified_name: TEST_DB.PUBLIC.TEST_ASSERTIONS_ALL_TIMES entity_schema: - col: col_date native_type: DATE # Field Value Assertion - type: field entity: urn:li:dataset:(urn:li:dataPlatform:snowflake,test_db.public.purchase_event,PROD) field: quantity condition: type: between min: 0 max: 10 schedule: type: on_table_change meta: entity_qualified_name: TEST_DB.PUBLIC.PURCHASE_EVENT entity_schema: - col: quantity native_type: FLOAT # Custom SQL Metric Assertion - type: sql entity: urn:li:dataset:(urn:li:dataPlatform:snowflake,test_db.public.purchase_event,PROD) statement: select mode(quantity) from test_db.public.purchase_event condition: type: equal_to value: 5 schedule: type: on_table_change meta: entity_qualified_name: TEST_DB.PUBLIC.PURCHASE_EVENT entity_schema: - col: quantity native_type: FLOAT

关键字段说明：

• version：配置文件格式版本，固定为1；namespace（内部别名id）标识该断言配置文件的唯一命名空间。此结构由 assertion_config_spec.py 中的AssertionsConfigSpec模型校验；
• entity：断言作用目标数据集的 URN，必须与 DataHub 中已摄取的 Snowflake 数据集一致；
• type：断言类型，取值为freshness/volume/field/sql；
• schedule：调度方式，支持cron（cron 表达式 + 时区）与on_table_change（表变更触发）；
• meta.entity_qualified_name：Snowflake 侧的完整表名（DB.SCHEMA.TABLE），用于生成 DMF 关联语句；
• meta.entity_schema：DMF 参数所需的一个列名与原生类型。Snowflake 不允许创建不带列参数的自定义数据指标函数，因此编译器需要从表 schema 中取任意一列作为ARGT TABLE(...)的参数（见 compiler.py 中get_dmf_args的实现）。

Step 2：用 DataHub CLI 注册断言

将断言注册到 DataHub，使其在 DataHub UI 中可见：

datahub assertions upsert -f examples/library/assertions_configuration.yml

upsert命令读取 YAML 文件，为每条断言生成一个urn:li:assertion:<id>并发射对应的AssertionInfoMCP 到 DataHub。其 CLI 实现在 assertions_cli.py。

注：仓库中该 assertions CLI 已标记为 deprecated（未来版本可能移除），并建议关注替代方案；编译与注册流程在当前版本仍然可用。

Step 3：用assertions compile编译为 Snowflake DMF

接下来使用compile命令生成可在 Snowflake 注册的 SQL 代码：

datahub assertions compile -f examples/library/assertions_configuration.yml -p snowflake -x DMF_SCHEMA=<db>.<schema-where-DMF-should-live>

命令参数：

DMF_SCHEMA是必需的扩展参数：编译器会严格校验其存在，否则报错Must specify value for DMF schema using -x DMF_SCHEMA=<db.schema>（见 compiler.py 的create方法）。

命令运行后会生成两个文件，默认位于target/目录（可用-o指定其他目录）：

• dmf_definitions.sql：将注册到 Snowflake 的 DMF 创建 SQL；
• dmf_associations.sql：将 DMF 与目标表关联、并配置调度计划的 SQL。

此外还会生成compile_report.json（编译报告），记录每条断言的编译成功/失败状态（见 assertions_cli.py）。

dmf_definitions.sql内容示例

该文件保存从 YAML 断言定义编译生成的 DMF 创建语句：

-- Example dmf_definitions.sql -- Start of Assertion 5c32eef47bd763fece7d21c7cbf6c659 CREATE or REPLACE DATA METRIC FUNCTION test_db.datahub_dmfs.datahub__5c32eef47bd763fece7d21c7cbf6c659 (ARGT TABLE(col_date DATE)) RETURNS NUMBER COMMENT = 'Created via DataHub for assertion urn:li:assertion:5c32eef47bd763fece7d21c7cbf6c659 of type volume' AS $$ select case when metric <= 1000 then 1 else 0 end from (select count(*) as metric from TEST_DB.PUBLIC.TEST_ASSERTIONS_ALL_TIMES ) $$; -- End of Assertion 5c32eef47bd763fece7d21c7cbf6c659 ....

要点解读：

• DMF 名称统一使用datahub__<assertion_id>前缀（源码见 compiler.py 的get_dmf_name），这个前缀在摄取阶段用于区分 DataHub 管理的 DMF 与外部 DMF；
• COMMENT中记录了断言 URN 与断言类型，便于溯源；
• 函数体是一个返回 0/1 的 CASE 表达式：metric <= 1000返回 1（通过），否则返回 0（失败）。所有类型断言的最终评估都会被统一折叠为 0/1 语义；
• 组装上述语句的模板见 dmf_generator.py 的create_dmf。

dmf_associations.sql内容示例

该文件保存将 DMF 与目标表关联、并配置调度计划的语句：

-- Example dmf_associations.sql -- Start of Assertion 5c32eef47bd763fece7d21c7cbf6c659 ALTER TABLE TEST_DB.PUBLIC.TEST_ASSERTIONS_ALL_TIMES SET DATA_METRIC_SCHEDULE = 'TRIGGER_ON_CHANGES'; ALTER TABLE TEST_DB.PUBLIC.TEST_ASSERTIONS_ALL_TIMES ADD DATA METRIC FUNCTION test_db.datahub_dmfs.datahub__5c32eef47bd763fece7d21c7cbf6c659 ON (col_date); -- End of Assertion 5c32eef47bd763fece7d21c7cbf6c659 ....

调度方式由断言 YAML 中的schedule决定，编译器会将其翻译为 Snowflake 的调度语法（见 compiler.py 的get_dmf_schedule）：

同时，编译器会校验同一张表上的所有断言必须使用完全相同的调度，否则抛错（见 compiler.py），这与下文"注意事项"中 Snowflake 的限制保持一致。

Step 4：在 Snowflake 环境中注册编译产物

将 Step 3 生成的两个 SQL 文件在 Snowflake 中执行。可以直接在 Snowflake UI 中运行，也可以使用 SnowSQL CLI：

snowsql -f dmf_definitions.sql snowsql -f dmf_associations.sql

:::note 在表上调度 Data Metric Function 会产生 SnowflakeServerless Credit 用量，计费细节参考 Snowflake 官方计费说明。若某条断言不再使用，请务必通过dmf_associations.sql中对应的方式 DROP 掉相关 Data Metric Function，避免持续产生费用。 :::

Step 5：运行摄取，把 DMF 结果回传到 DataHub

DMF 注册完成后，Snowflake 会在目标表更新时（TRIGGER_ON_CHANGES）或按固定计划自动执行它们。

要把生成的数据质量断言结果回传到 DataHub，需要以特殊配置标志运行 DataHub 摄取：include_assertion_results: true：

# Your DataHub Snowflake Recipe source: type: snowflake config: # ... include_assertion_results: True # ...

然后照常执行摄取：

datahub ingest -c snowflake.yml

摄取过程中，DataHub 会查询 Snowflake 中存储的最新 DMF 结果，将其转换为 DataHub Assertion Results，并在摄取期间（通过 CLI 或 UI）以普通断言的形式报告回来。

从源码看，摄取器在 snowflake_assertion.py 中查询 snowflake_query.py 定义的dmf_assertion_resultsSQL，从SNOWFLAKE.LOCAL.DATA_QUALITY_MONITORING_RESULTS视图读取结果，并按MEASUREMENT_TIME时间窗口（对应 recipe 中的start_time/end_time）过滤：

• 默认情况下（include_externally_managed_dmfs=False），查询会通过METRIC_NAME ilike 'datahub\_\_%' escape '\'过滤，只取 DataHub 创建的 DMF；
• 结果行中VALUE被映射为断言运行状态：VALUE=1→ SUCCESS（通过），VALUE=0→ FAILURE（失败），其他值 → ERROR（见 snowflake_assertion.py）；
• 每条结果生成一个AssertionRunEvent工作单元，并携带MEASUREMENT_TIME作为时间戳，DataHub UI 据此绘制历史时间线。

单元测试 test_snowflake_assertion.py 验证了上述查询行为：默认查询包含datahub前缀过滤，开启外部 DMF 后则不附加任何ilike过滤。

六、摄取外部（用户自建）DMF

除了 DataHub 创建的 DMF，你还可以摄取自己用 Snowflake 原生方式创建的 Data Metric Function 的结果。"外部"指那些未经过 DataHub assertion compiler、直接创建在 Snowflake 中的 DMF——它们独立于 DataHub 的管理范围之外。

6.1 为什么要摄取外部 DMF

• 存量 DMF：在采用 DataHub 之前就已经存在于 Snowflake 的 DMF，希望在 DataHub 中看到其结果而无需重建；
• 自定义逻辑：需要 DataHub assertion compiler 不支持的 DMF 逻辑（例如复杂的多表检查）；
• 团队工作流：不同团队直接在 Snowflake 中管理 DMF，但你希望在 DataHub 中获得集中可见性；
• 渐进式采用：在完全迁移到 DataHub 管理的断言之前，先在 DataHub 中监控现有的数据质量检查。

6.2 启用外部 DMF 摄取

在 Snowflake recipe 中添加include_externally_managed_dmfs标志：

source: type: snowflake config: # ... connection config ... # 启用断言结果摄取（必需） include_assertion_results: true # 启用外部 DMF 摄取（新增） include_externally_managed_dmfs: true # 断言结果的时间窗口 start_time: "-7 days"

两个标志必须同时开启外部 DMF 摄取才能生效。这一约束在配置模型层被强制校验：include_externally_managed_dmfs: True而include_assertion_results未开启时，配置校验会直接抛出ValueError（见 snowflake_config.py）。

6.3 外部 DMF 的要求

外部 DMF 必须以1表示 SUCCESS、以0表示 FAILURE。

DataHub 将 SnowflakeDATA_QUALITY_MONITORING_RESULTS表中的VALUE列解释为：

• VALUE = 1→ 断言PASSED
• VALUE = 0→ 断言FAILED

这是因为 DataHub 无法解释任意的返回值（例如 "100 个空值行"——这到底是好还是坏？）。你必须把通过/失败逻辑构建到 DMF 本身中。

:::warning 如果我的 DMF 返回其他值怎么办？ 如果 DMF 返回 0 或 1 以外的值，DataHub 会把断言结果标记为ERROR：

• VALUE = 1→PASSED
• VALUE = 0→FAILED
• VALUE != 0 且 VALUE != 1（如 5、100、-1）→ERROR

ERROR 状态表示该 DMF 未按 DataHub 摄取的要求正确配置。可通过以下方式识别这些情况：

1. 检查摄取日志中的警告，例如：DMF 'my_dmf' returned invalid value 100. Expected 1 (pass) or 0 (fail). Marking as ERROR.
2. 在 DataHub UI 中查找处于 ERROR 状态的断言 :::

这一映射逻辑在 snowflake_assertion.py 中有直接实现，日志中的警告文案与文档描述一致。

示例：正确编写外部 DMF

错误示例—— 返回原始计数（DataHub 无法解释）：

CREATE DATA METRIC FUNCTION my_null_check(ARGT TABLE(col VARCHAR)) RETURNS NUMBER AS $$ SELECT COUNT(*) FROM ARGT WHERE col IS NULL $$; -- 返回：0, 5, 100 等 —— DataHub 无法判定通过/失败！

正确示例—— 返回 1（通过）或 0（失败）：

CREATE DATA METRIC FUNCTION my_null_check(ARGT TABLE(col VARCHAR)) RETURNS NUMBER AS $$ SELECT CASE WHEN COUNT(*) = 0 THEN 1 ELSE 0 END FROM ARGT WHERE col IS NULL $$; -- 返回：无空值时返回 1（通过），存在空值时返回 0（失败）

正确示例—— 带阈值：

CREATE DATA METRIC FUNCTION my_null_check_threshold(ARGT TABLE(col VARCHAR)) RETURNS NUMBER AS $$ SELECT CASE WHEN COUNT(*) <= 10 THEN 1 ELSE 0 END FROM ARGT WHERE col IS NULL $$; -- 返回：空值 ≤10 时返回 1（通过），空值 >10 时返回 0（失败）

6.4 外部 DMF 与 DataHub 创建 DMF 的差异

从源码看，外部 DMF 的 URN 通过SnowflakeExternalDmfKey基于REFERENCE_ID（DMF-表-列关联的唯一标识）与可选platform_instance生成确定性 GUID（见 snowflake_assertion.py 与 测试用例），从而保证同一REFERENCE_ID多次摄取生成相同的断言 URN、不同REFERENCE_ID互不冲突。而对于datahub__前缀的 DMF，则直接从名称中提取 GUID。

6.5 外部 DMF 在 DataHub UI 中的呈现

外部 DMF 在 DataHub 中以如下形式出现：

• Assertion Type：CUSTOM
• Source：EXTERNAL
• Platform Instance：Snowflake platform instance（若已配置）
• Description："External Snowflake DMF: {dmf_name}"
• Custom Properties：
  • snowflake_dmf_name：DMF 函数名
  • snowflake_reference_id：Snowflake 对 DMF-表绑定的唯一标识
  • snowflake_dmf_columns：DMF 作用的列列表（逗号分隔）

这些字段在 snowflake_assertion.py 的_create_assertion_info_workunit中被写入AssertionInfo的customAssertion与customProperties；单列 DMF 还会生成对应的 schema field URN，从而把断言挂到具体列上。

你可以在 DataHub UI 中关联数据集的Quality标签页查看外部 DMF 断言，它们会与 DataHub 创建的断言一同展示通过/失败历史。

七、注意事项（Caveats）

以下限制源自 Snowflake 平台本身，使用前需知晓：

• 目前 Snowflake 最多支持1000 个 DMF-表关联，因此你无法为 Snowflake 定义超过 1000 条断言；
• 目前 Snowflake不允许在 DMF 定义中使用 JOIN 查询或非确定性函数，因此 SQL 断言或过滤器部分不能使用这些特性；
• 目前一张表上调度的所有 DMF必须遵循完全相同的调度计划，因此不能为同一张表上的不同断言设置不同的调度；
• 目前 DMF仅支持常规表，不支持动态表（dynamic tables）或外部表（external tables）。

其中"同一表必须同调度"的限制在编译器侧已被强制执行（见 compiler.py），"仅支持常规表"也与 Snowflake 摄取配置中table_types的取值（BASE TABLE/EXTERNAL TABLE）相互印证（见 snowflake_config.py）。

八、FAQ

原文档中该章节标注为 "Coming soon!"，暂未提供具体问答内容。如果你在使用中遇到问题，可以结合上文提到的源码路径（编译器、摄取器、查询与测试）定位行为细节，或在 DataHub 官方渠道寻求支持。

九、总结与后续探索

本文完整走通了"YAML 声明断言 →assertions upsert注册 →assertions compile编译 → Snowflake 注册与调度 →include_assertion_results摄取结果"的 Snowflake DMF 断言全链路，并说明了外部 DMF 的接入要求与 0/1 返回值约定。想深入了解的读者可以继续在仓库中探索：

• 编译产物生成与调度翻译：compiler.py、dmf_generator.py、metric_sql_generator.py；
• 断言结果摄取与外部 DMF 处理：snowflake_assertion.py、snowflake_query.py；
• 摄取配置项定义与校验：snowflake_config.py；
• 单元测试：test_snowflake_assertion.py；
• YAML 断言定义示例：assertions_configuration.yml；
• 四种断言类型的详细概念：Freshness、Volume、Column、Custom SQL；
• Snowflake 摄取入门：overview、setup、configuration。

【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub