DataHub Python SDK 批量创建带异常检测Anomaly Detection的数据质量断言实战指南【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub本篇技术指南讲解如何基于 DataHub Cloud Python SDKacryl-datahub-cloud在 DataHub 中以编程方式批量创建启用 Anomaly Detection原称 Smart Assertions的数据质量断言覆盖表级与列级两类断言、表的发现与筛选、断言 URN 的存储与更新以及面向生产环境的批处理与最佳实践。读完本文你将掌握在数百张表、数千个列上规模化落地 Freshness / Volume / Column Metric / Custom SQL 四类智能断言的一整套可运行方案并了解其底层实现原理。背景什么是带 Anomaly Detection 的断言在进入批量创建之前先理解 Anomaly Detection 的本质。根据 异常检测官方文档Anomaly Detection 是可选的断言能力用 AI 驱动的动态阈值替代固定阈值你不再需要手工指定行数必须介于 1000 到 2000而是由模型学习底层指标的历史规律趋势、季节性、典型波动当最新取值落在正常范围之外时触发告警。Anomaly Detection 当前支持四类断言成熟度各不相同断言类型Anomaly Detection 阶段关键约束Freshness 断言GA基于仓库查询或 ingestion 信号配合 DataHuboperationaspect 可用于任何上报 Operations 的平台含 Clickhouse、Oracle、Dremio 等Volume 断言GA基于仓库查询或 Dataset Profile配合 Dataset Profile 可用于任何上报行数 profile 的平台含 Iceberg、Postgres、MySQL 等Column Metric 断言Public Beta仅支持null_count、unique_count、empty_count、zero_count、negative_count指标Custom SQL 断言Public Beta需要活跃的仓库连接仅限 Snowflake / Redshift / BigQuery / Databricks需要特别注意的是Column Value 断言如值匹配正则值属于集合和 Schema 断言不支持 Anomaly Detection前者是确定性校验、没有统计意义上的正常后者是离散事件而非异常。另外新创建的 Anomaly Detection 断言会先收集14 天历史数据才正式开始告警——期间照常调度评估并记录结果但不报失败若想跳过等待期可在创建时启用 Backfill Assertion History从仓库回填指标历史让预测从第一天起就可用。为什么需要批量创建断言通过 Python SDK 批量创建断言的核心价值在于规模化数据质量跨数百甚至数千张表应用一致的断言策略自动化断言管理基于元数据模式标签、域名、平台、命名规律以编程方式创建与更新断言落地治理策略确保所有关键表都有恰当的数据质量检查节省时间避免在 UI 中逐条手工创建。前置条件批量创建前需要满足以下条件安装 DataHub Cloud Python SDKpip install acryl-datahub-cloud需要说明的是sync_smart_*_assertion这一组 API 属于 Cloud SDK 的扩展能力。在当前开源仓库的 assertion_client.py 中AssertionClient.__getattr__的实现明确说明未安装acryl-datahub-cloud时client.assertions上任何 Cloud-only 方法都会抛出SdkUsageError并提示先用pip install acryl-datahub-cloud安装。也就是说仅安装开源acryl-datahub时只能使用sync_custom_assertion与report_assertion_result可参考 sync_custom_assertion.py 示例批量智能断言必须安装 Cloud SDK。配置有效的 DataHub Cloud 凭证server URL 与具备相应权限的 access token发起调用的 actor 必须对目标数据集拥有Edit Assertions与Edit Monitors权限目标数据集必须已存在于 DataHub 实例中。如果尝试为不存在的实体创建断言GMS 会持续向日志上报错误。本指南目标本指南将演示如何用 DataHub Cloud Python SDK 以编程方式创建大量启用 Anomaly Detection 的断言。整体流程概览批量断言创建的完整流程分七步发现表通过搜索或直接指定表 URN 找到目标数据集创建表级断言为每张表添加 Freshness 与 Volume 断言获取列信息读取每张表的 schema 细节创建列级断言为相关列添加 Column Metric 断言创建订阅为数据集或断言创建订阅以便接收变更通知存储断言 URN保存断言标识符供后续更新更新已有断言基于存储的 URN 对断言参数做增量调整。环境初始化连接 DataHubfrom datahub.sdk import DataHubClient client DataHubClient(serveryour_server, tokenyour_token)参数说明serverDataHub GMS 服务器地址本地http://localhost:8080托管hostedhttps://your_datahub_url/gmstoken需先从 DataHub 实例中生成个人访问令牌Personal Access Token。也可以先设置DATAHUB_GMS_URL、DATAHUB_GMS_TOKEN环境变量或运行datahub init生成~/.datahubenv文件再通过from_env()初始化from datahub.sdk import DataHubClient client DataHubClient.from_env()并行处理的重要注意事项对同一数据集的批量断言创建务必在单线程中执行避免竞态条件race conditions对同一数据集的订阅 API 调用也必须在单线程中执行如果直接订阅断言请确保脚本按数据集维度单线程运行。Step 1发现目标表方案 A显式指定表 URNfrom datahub.metadata.urns import DatasetUrn # Define specific tables you want to add assertions to table_urns [ urn:li:dataset:(urn:li:dataPlatform:snowflake,database.schema.users,PROD), urn:li:dataset:(urn:li:dataPlatform:snowflake,database.schema.orders,PROD), urn:li:dataset:(urn:li:dataPlatform:snowflake,database.schema.products,PROD), ] # Convert to DatasetUrn objects datasets [DatasetUrn.from_string(urn) for urn in table_urns]方案 B按命名模式搜索表更全面的搜索能力与筛选选项参见 Search API 文档from datahub.sdk.search_filters import FilterDsl from datahub.metadata.urns import DatasetUrn # Search for tables matching criteria def find_tables_by_pattern(client, platformsnowflake, name_patternproduction_*): Find tables matching a specific pattern. # Create filters for datasets on a specific platform with name pattern filters FilterDsl.and_( FilterDsl.entity_type(dataset), FilterDsl.platform(platform), FilterDsl.custom_filter(name, EQUAL, [name_pattern]) ) # Use the search client to find matching datasets urns list(client.search.get_urns(filterfilters)) return [DatasetUrn.from_string(str(urn)) for urn in urns] # Use the search function datasets find_tables_by_pattern(client, platformsnowflake, name_patternproduction_*)从源码看search_filters.py 中的FilterDsl提供了and_/or_/not_、entity_type、entity_subtype、platform、domain、container、env、owner、glossary_term、tag、has_custom_property、soft_deleted、custom_filter等静态工厂方法可自由组合出精确的搜索谓词。其中FilterDsl.and_在编译阶段通过笛卡尔积合并各子句见_And.compile的实现 search_filters.py例如(A or B) and (C or D)会被展开为(A and C) or (A and D) or (B and C) or (B and D)因此理论上可表达任意复杂度的布尔查询。方案 C按标签或域名获取表def find_tables_by_tag(client, tag_namecritical): Find tables with a specific tag. # Create filters for datasets with a specific tag filters FilterDsl.and_( FilterDsl.entity_type(dataset), FilterDsl.custom_filter(tags, EQUAL, [furn:li:tag:{tag_name}]) ) # Use the search client to find matching datasets urns list(client.search.get_urns(filterfilters)) return [DatasetUrn.from_string(str(urn)) for urn in urns] # Find all tables tagged as critical critical_datasets find_tables_by_tag(client, critical)补充说明源码中的FilterDsl.tag()工厂方法search_filters.py内部使用_TagFilter其校验器要求传入的必须是urn:li:tag:前缀的合法 tag URN见 search_filters.py因此上例中手写furn:li:tag:{tag_name}的拼接方式与FilterDsl.tag(urn:li:tag:critical)等价。此外custom_filter的condition参数支持EQUAL、CONTAIN、START_WITH、END_WITH、GREATER_THAN、LESS_THAN等条件可用于数值或时间戳字段的范围过滤。Step 2创建表级断言创建断言前先准备一个 URN 注册表用于存放后续创建出的断言标识符# Storage for assertion URNs (for later updates) assertion_registry { freshness: {}, volume: {}, smart_sql: {}, column_metrics: {} }Freshness 断言 Anomaly Detectiondef create_freshness_assertions(datasets, client, registry): Create Freshness assertions with Anomaly Detection for multiple datasets. for dataset_urn in datasets: try: freshness_assertion client.assertions.sync_smart_freshness_assertion( dataset_urndataset_urn, display_namefFreshness Anomaly Monitor, # Detection mechanism - information_schema is recommended detection_mechanisminformation_schema, # AI sensitivity setting sensitivitymedium, # options: low, medium, high # Tags for grouping (supports urns or plain tag names!) tags[automated, freshness, data_quality], # Enable the assertion enabledTrue ) # Store the assertion URN for future reference registry[freshness][str(dataset_urn)] str(freshness_assertion.urn) print(f✅ Created freshness assertion for {dataset_urn.name}: {freshness_assertion.urn}) except Exception as e: print(f❌ Failed to create freshness assertion for {dataset_urn.name}: {e}) # Create freshness assertions for all datasets create_freshness_assertions(datasets, client, assertion_registry)Volume 断言 Anomaly Detectiondef create_volume_assertions(datasets, client, registry): Create Volume assertions with Anomaly Detection for multiple datasets. for dataset_urn in datasets: try: volume_assertion client.assertions.sync_smart_volume_assertion( dataset_urndataset_urn, display_namefVolume Anomaly Monitor, # Detection mechanism options detection_mechanisminformation_schema, # AI sensitivity setting sensitivitymedium, # Tags for grouping tags[automated, volume, data_quality], # Schedule (optional - defaults to hourly) schedule0 */6 * * *, # Every 6 hours # Enable the assertion enabledTrue ) # Store the assertion URN registry[volume][str(dataset_urn)] str(volume_assertion.urn) print(f✅ Created volume assertion for {dataset_urn.name}: {volume_assertion.urn}) except Exception as e: print(f❌ Failed to create volume assertion for {dataset_urn.name}: {e}) # Create volume assertions for all datasets create_volume_assertions(datasets, client, assertion_registry)Custom SQL 断言 Anomaly DetectionPublic Betadef create_smart_sql_assertions(datasets, client, registry): Create Custom SQL assertions with Anomaly Detection for multiple datasets. # Define SQL queries to run on each table sql_queries { row_count: SELECT COUNT(*) FROM {table_name}, null_check: SELECT COUNT(*) FROM {table_name} WHERE id IS NULL, active_records: SELECT COUNT(*) FROM {table_name} WHERE status active, } for dataset_urn in datasets: registry[smart_sql][str(dataset_urn)] {} for query_name, query_template in sql_queries.items(): try: table_name dataset_urn.name statement query_template.format(table_nametable_name) sql_assertion client.assertions.sync_smart_sql_assertion( dataset_urndataset_urn, display_namefSQL Anomaly Monitor - {query_name}, statementstatement, # AI-powered sensitivity setting sensitivitymedium, # options: low, medium, high # Tags for grouping tags[automated, anomaly_detection, query_name], # Schedule schedule0 */6 * * *, # Every 6 hours # Enable the assertion enabledTrue ) registry[smart_sql][str(dataset_urn)][query_name] str(sql_assertion.urn) print(f✅ Created Custom SQL anomaly monitor {query_name} for {dataset_urn.name}: {sql_assertion.urn}) except Exception as e: print(f❌ Failed to create Custom SQL anomaly monitor {query_name} for {dataset_urn.name}: {e}) # Create Custom SQL anomaly monitors for all datasets create_smart_sql_assertions(datasets, client, assertion_registry)关键参数说明sensitivity灵敏度low/medium/high三档。灵敏度越高模型对数据的拟合越紧、越容易触发告警越低则容忍更大的数据波动。这对应 异常检测文档 中Tuning一节所述的灵敏度调优手段。detection_mechanism检测机制。information_schema表示通过仓库的 information_schema 查询信号列级断言中还会见到all_rows_query_datahub_dataset_profile基于 DataHub Dataset Profile 信号。schedulecron 表达式调度可选项Volume 与 Custom SQL 默认每小时一次。tags支持传入 URN 或普通标签名——普通标签名会自动转换为urn:li:tag:name形式这是本指南末尾会重点强调的易用性特性。Step 3获取列信息要为列创建断言必须先读取数据集的 schemadef get_dataset_columns(client, dataset_urn): Get column information for a dataset. try: # Get dataset using the entities client dataset client.entities.get(dataset_urn) if dataset and hasattr(dataset, schema) and dataset.schema: return [ { name: field.field_path, type: field.native_data_type, nullable: field.nullable if hasattr(field, nullable) else True } for field in dataset.schema.fields ] return [] except Exception as e: print(f❌ Failed to get columns for {dataset_urn}: {e}) return [] # Get columns for each dataset dataset_columns {} for dataset_urn in datasets: columns get_dataset_columns(client, dataset_urn) dataset_columns[str(dataset_urn)] columns print(f Found {len(columns)} columns in {dataset_urn.name})这里通过client.entities.get(dataset_urn)拉取实体从dataset.schema.fields中读取每个字段的field_path列名、native_data_type原生数据类型如VARCHAR、INTEGER与nullable标记。列的类型信息是下一步按规则筛选列的关键输入。Step 4创建列级断言Column Metric 断言 Anomaly DetectionPublic Betadef create_column_assertions(datasets, columns_dict, client, registry): Create Column Metric assertions with Anomaly Detection for multiple datasets and columns. # Define rules for which columns should get which assertions assertion_rules { # Null count checks for critical columns null_checks: { column_patterns: [id, *_id, user_id, email], metric_type: null_count, }, # Unique count checks for ID columns unique_checks: { column_patterns: [*_id, email, username], metric_type: unique_count, }, # Empty count checks for string columns empty_checks: { column_patterns: [name, description, title], metric_type: empty_count, }, } for dataset_urn in datasets: dataset_key str(dataset_urn) columns columns_dict.get(dataset_key, []) if not columns: print(f⚠️ No columns found for {dataset_urn.name}) continue registry[column_metrics][dataset_key] {} for column in columns: column_name column[name] column_type column[type].upper() # Apply assertion rules based on column name and type for rule_name, rule_config in assertion_rules.items(): if should_apply_rule(column_name, column_type, rule_config): try: assertion client.assertions.sync_smart_column_metric_assertion( dataset_urndataset_urn, column_namecolumn_name, metric_typerule_config[metric_type], display_namef{rule_name.replace(_, ).title()} - {column_name}, # Detection mechanism for column metrics detection_mechanismall_rows_query_datahub_dataset_profile, # Tags (plain names automatically converted to URNs) tags[automated, column_quality, rule_name], enabledTrue ) # Store assertion URN if column_name not in registry[column_metrics][dataset_key]: registry[column_metrics][dataset_key][column_name] {} registry[column_metrics][dataset_key][column_name][rule_name] str(assertion.urn) print(f✅ Created {rule_name} assertion for {dataset_urn.name}.{column_name}) except Exception as e: print(f❌ Failed to create {rule_name} assertion for {dataset_urn.name}.{column_name}: {e}) def should_apply_rule(column_name, column_type, rule_config): Determine if a rule should be applied to a column. import fnmatch # Check column name patterns for pattern in rule_config[column_patterns]: if fnmatch.fnmatch(column_name.lower(), pattern.lower()): return True # Add type-based rules if needed if rule_config.get(column_types): return any(col_type in column_type for col_type in rule_config[column_types]) return False # Create column assertions create_column_assertions(datasets, dataset_columns, client, assertion_registry)这段代码演示了一个典型的规则引擎式列筛选assertion_rules定义了按列名通配符fnmatch匹配的规则集should_apply_rule决定某列是否命中规则命中后调用sync_smart_column_metric_assertion创建断言。如需扩展可在rule_config中增加column_types键来按数据类型如VARCHAR、NUMERIC补充筛选。Step 5创建订阅关于如何在数据集或断言上创建订阅参见订阅 SDK 教程。注意批量创建订阅时必须单线程执行以避免竞态条件。另外强烈建议在数据集级别创建订阅而不是为单个断言逐一创建订阅这样后续的持续管理会简单得多。Step 6存储断言 URN断言创建完成后建议将 URN 注册表持久化到文件便于未来更新与审计。保存到文件import json from datetime import datetime def save_assertion_registry(registry, filenameNone): Save assertion URNs to a file for future reference. if filename is None: timestamp datetime.now().strftime(%Y%m%d_%H%M%S) filename fassertion_registry_{timestamp}.json # Add metadata registry_with_metadata { created_at: datetime.now().isoformat(), total_assertions: { freshness: len(registry[freshness]), volume: len(registry[volume]), column_metrics: sum( len(cols) for cols in registry[column_metrics].values() ) }, assertions: registry } with open(filename, w) as f: json.dump(registry_with_metadata, f, indent2) print(f Saved assertion registry to {filename}) return filename # Save the registry registry_file save_assertion_registry(assertion_registry)从文件加载用于后续更新def load_assertion_registry(filename): Load assertion URNs from a previously saved file. with open(filename, r) as f: data json.load(f) return data[assertions] # Later, load for updates # assertion_registry load_assertion_registry(assertion_registry_20240101_120000.json)Step 7更新已有断言由于sync_smart_*_assertion系列是同步sync语义传入已存在的urn即可原地更新无需先删除再创建def update_existing_assertions(registry, client): Update existing assertions using stored URNs. # Update freshness assertions for dataset_urn, assertion_urn in registry[freshness].items(): try: updated_assertion client.assertions.sync_smart_freshness_assertion( dataset_urndataset_urn, urnassertion_urn, # Provide existing URN for updates # Update any parameters as needed sensitivityhigh, # Change sensitivity tags[automated, freshness, data_quality, updated], enabledTrue ) print(f Updated freshness assertion {assertion_urn}) except Exception as e: print(f❌ Failed to update freshness assertion {assertion_urn}: {e}) # Update assertions when needed # update_existing_assertions(assertion_registry, client)高级模式模式一基于元数据条件的条件式断言创建可以结合实体的标签、属性等元数据对不同类型的表施加差异化策略——例如对打了critical标签的表使用更高灵敏度def create_conditional_assertions(datasets, client): Create assertions based on dataset metadata conditions. for dataset_urn in datasets: try: # Get dataset metadata dataset client.entities.get(dataset_urn) # Check if dataset has specific tags if dataset.tags and any(critical in str(tag.tag) for tag in dataset.tags): # Create more stringent assertions for critical datasets client.assertions.sync_smart_freshness_assertion( dataset_urndataset_urn, sensitivityhigh, detection_mechanisminformation_schema, tags[critical, automated, freshness] ) # Check dataset size and apply appropriate volume checks if dataset.dataset_properties: # Create different volume assertions based on table characteristics pass except Exception as e: print(f❌ Error processing {dataset_urn}: {e})模式二带错误处理与速率限制的批处理面对大规模数据集建议分批提交并在批次间加入延时同时收集成功/失败明细import time from typing import List, Dict, Any def batch_create_assertions( datasets: List[DatasetUrn], client: DataHubClient, batch_size: int 10, delay_seconds: float 1.0 ) - Dict[str, Any]: Create assertions in batches with error handling and rate limiting. results { successful: [], failed: [], total_processed: 0 } for i in range(0, len(datasets), batch_size): batch datasets[i:i batch_size] print(fProcessing batch {i//batch_size 1}: {len(batch)} datasets) for dataset_urn in batch: try: # Create assertion assertion client.assertions.sync_smart_freshness_assertion( dataset_urndataset_urn, tags[batch_created, automated], enabledTrue ) results[successful].append({ dataset_urn: str(dataset_urn), assertion_urn: str(assertion.urn) }) except Exception as e: results[failed].append({ dataset_urn: str(dataset_urn), error: str(e) }) results[total_processed] 1 # Rate limiting between batches if i batch_size len(datasets): time.sleep(delay_seconds) return results # Use batch processing batch_results batch_create_assertions(datasets, client, batch_size5) print(fBatch results: {batch_results[total_processed]} processed, f{len(batch_results[successful])} successful, f{len(batch_results[failed])} failed)最佳实践1. 标签策略使用一致的标签名对断言分组例如[automated, freshness, critical]普通标签名会自动转换为 URNmy_tag→urn:li:tag:my_tag无需手工拼 URN为不同类型的断言与优先级建立标签层级便于后续按标签批量检索、订阅或审计。2. 错误处理始终用 try-catch 包裹断言创建逻辑将失败信息记录下来供事后排查为瞬时故障实现重试逻辑。3. URN 管理将断言 URN 存放到持久化位置文件、数据库等文件名使用带时间戳的有意义命名记录断言创建的时间与原因等元数据。4. 性能考量后台架构面向大规模操作设计但写入是异步提交到 Kafka 队列的大规模操作可能存在明显延迟。若遇到问题可参考以下建议错峰执行避免大批量操作造成 Kafka lag 尖峰重跑 sync 前等待更新前先等 GMS 完成上一轮处理通过检查最近一条数据是否已反映在 GMS 中来避免不一致与重复监控处理状态通过 DataHub UI 或 API 确认操作全部完成分批处理数据集避免一次性压垮 API必要时在批次间加入延时。5. 测试策略先用一小部分数据集做试点在大规模处理前先验证断言创建是否正常用已存在的断言测试更新场景。完整示例脚本以下脚本把上述步骤串成一个可直接参考的端到端流程实际运行时请先补齐create_freshness_assertions、create_volume_assertions、get_dataset_columns、create_column_assertions、save_assertion_registry等上文定义的函数#!/usr/bin/env python3 Complete example script for bulk creating assertions with Anomaly Detection enabled. import json import time from datetime import datetime from typing import List, Dict, Any from datahub.sdk import DataHubClient from datahub.ingestion.graph.client import DataHubGraph from datahub.metadata.urns import DatasetUrn def main(): # Initialize the DataHub client client DataHubClient( serverhttps://your-datahub-instance.com, tokenyour-access-token, ) # The client provides both search and entity access # Define target datasets table_urns [ urn:li:dataset:(urn:li:dataPlatform:snowflake,prod.analytics.users,PROD), urn:li:dataset:(urn:li:dataPlatform:snowflake,prod.analytics.orders,PROD), urn:li:dataset:(urn:li:dataPlatform:snowflake,prod.analytics.products,PROD), ] datasets [DatasetUrn.from_string(urn) for urn in table_urns] # Registry to store assertion URNs assertion_registry { freshness: {}, volume: {}, column_metrics: {} } print(f Starting bulk assertion creation for {len(datasets)} datasets) # Step 1: Create table-level assertions print(\n Creating freshness assertions...) create_freshness_assertions(datasets, client, assertion_registry) print(\n Creating volume assertions...) create_volume_assertions(datasets, client, assertion_registry) # Step 2: Get column information and create column assertions print(\n Analyzing columns and creating column assertions...) dataset_columns {} for dataset_urn in datasets: columns get_dataset_columns(client, dataset_urn) dataset_columns[str(dataset_urn)] columns create_column_assertions(datasets, dataset_columns, client, assertion_registry) # Step 3: Save results print(\n Saving assertion registry...) registry_file save_assertion_registry(assertion_registry) # Summary total_assertions ( len(assertion_registry[freshness]) len(assertion_registry[volume]) sum(len(cols) for cols in assertion_registry[column_metrics].values()) ) print(f\n✅ Bulk assertion creation complete!) print(f Total assertions created: {total_assertions}) print(f Freshness assertions: {len(assertion_registry[freshness])}) print(f Volume assertions: {len(assertion_registry[volume])}) print(f Column assertions: {sum(len(cols) for cols in assertion_registry[column_metrics].values())}) print(f Registry saved to: {registry_file}) if __name__ __main__: main()延伸阅读与底层线索理解 Anomaly Detection 的完整机制14 天学习期、灵敏度调优、训练数据回看窗口、异常反馈、时间序列分桶等见异常检测文档及其子页面Freshness 断言、Volume 断言、Column Metric 断言、Custom SQL 断言、Backfill Assertion History搜索客户端的完整过滤能力F.platform、F.env、F.entity_type、F.domain、F.soft_deleted、F.has_custom_property以及and_/or_/not_逻辑组合见Search API 文档与 search_filters.py在数据集或断言上创建订阅的用法与可用变更类型见订阅教程开源 SDK 自带的 CUSTOM 断言上报示例用于外部监控工具自报数据质量见 sync_custom_assertion.py。综上借助 DataHub Cloud Python SDK 的sync_smart_*_assertion系列接口配合搜索筛选、规则化列匹配、URN 注册表持久化与批处理节奏控制即可在规模化场景下稳定地落地带 Anomaly Detection 的智能数据质量监控。其中标签名的自动 URN 转换特性普通标签名 →urn:li:tag:name进一步简化了断言的分类组织与后续管理。【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考