DataHub Cloud v0.3.3 版本技术解析按需断言、Slack 事件协作与断言级订阅【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub2024 年 6 月 25 日发布的 DataHub Cloud v0.3.3 是数据质量Observe能力的一次集中增强它首次支持从 UI 中按需触发断言运行、为 Slack 中的事件Incident协作提供了更丰富的信息载体并引入断言级订阅以精确控制质量通知的粒度。本文以官方发布说明为主体结合仓库中的 Assertions API 教程、Incidents 指南、Subscriptions 教程 及 SDK 示例源码逐项拆解这些新能力的概念、配置方式与底层调用方式帮助读者在升级后第一时间用上这些功能。版本概览与升级前提发布信息项目内容版本号v0.3.3发布可用日期2024-06-25推荐 CLI/SDKv0.13.3面向平台DataHub CloudSaaS强烈建议升级 CLI/SDK 至 v0.13.3发布说明中对 CLI/SDK 给出了明确的升级要求凡是使用 DataHub CLI/SDK 的场景——包括终端命令行、GitHub Actions、Airflow 调度、Python SDK、Java SDK 等——都强烈建议升级到 v0.13.3。官方给出的理由是 CLI 中持续在推送修复升级到统一版本也有助于平台侧提供更好的支持。对应到本仓库CLI 相关入口与用法可参考 docs/cli.mdSDK 侧的断言与订阅调用示例则存放在 metadata-ingestion/examples/library/下文会逐一引用。新能力一按需运行断言On-Demand Assertionsv0.3.3 的核心变化之一是允许用户从 DataHub Cloud UI 直接触发断言的按需运行而不必等待既定调度周期。该功能默认不开放需要联系 DataHub 团队在账号侧开启发布说明原文Reach out to DataHub team to enable this feature。断言运行的两种评估模式理解按需运行前先明确 DataHub Cloud 中断言是如何被评估的。根据 docs/managed-datahub/observe/assertions.md 的说明断言有两种评估方式Active query主动查询DataHub Cloud 按调度周期直接对数据源执行 SQL 查询。适用于 Snowflake、Redshift、BigQuery、Databricks。Ingestion-driven摄取驱动DataHub Cloud 基于摄取阶段已上报的元数据Operation操作事件、DatasetProfile数据集画像、已摄取 Schema进行评估适用于任何平台但评估节奏受摄取节奏约束。按需运行适用于 DataHub Cloud 原生执行的断言发布说明中同时提醒若断言是由外部工具3rd party runner执行的则相关 API 会返回错误。GraphQL 层三个按需运行 MutationAssertions API 教程 中提供了三个按需运行入口共同约定如下saveResult/saveResults是否将运行结果写入 DataHub 后端UI 可见。默认true设为false则结果不落库。async是否异步运行。默认false同步执行异步模式true下 API 立即返回结果需通过assertion(urn: String!)查询的runEvents字段获取。同步超时限制同步运行单条断言目前最长30 秒。1. 运行单条断言runAssertionmutation runAssertion { runAssertion(urn: urn:li:assertion:your-assertion-id, saveResult: true) { type nativeResults { key value } } }其中type返回断言运行结果取值为SUCCESS、FAILURE或ERROR。成功的响应示例{ data: { runAssertion: { type: SUCCESS, nativeResults: [ { key: Value, value: 1382 } ] } } }2. 运行一组断言runAssertionsmutation runAssertions { runAssertions( urns: [ urn:li:assertion:your-assertion-id-1 urn:li:assertion:your-assertion-id-2 ] saveResults: true ) { passingCount failingCount errorCount results { urn result { type nativeResults { key value } } } } }响应中为每个 URN 返回一个结果对象外部执行的断言会被静默地从结果集中省略。3. 运行某数据资产的全部断言runAssertionsForAssetmutation runAssertionsForAsset { runAssertionsForAsset( urn: urn:li:dataset:(urn:li:dataPlatform:snowflake,purchase_events,PROD) saveResults: true ) { passingCount failingCount errorCount results { urn result { type nativeResults { key value } } } } }进阶按标签运行 动态参数按标签筛选运行先通过addTagmutation 给断言打标签resourceUrn指向断言 URN再在runAssertionsForAsset中传入tagUrns参数即可只运行带特定标签的断言子集适合按重要程度分组执行。动态参数在断言的 SQL 片段中写入${parameterName}占位符运行时可传parameters: [{ key: parameterName, value: parameterValue }]动态注入。这对阈值随业务时段变化的场景如按一天中不同时段调整限额非常有用。Python SDK 侧仓库中的真实示例仓库 metadata-ingestion/examples/library/run_assertion.py 展示了最简调用方式from datahub.ingestion.graph.client import DatahubClientConfig, DataHubGraph graph DataHubGraph(configDatahubClientConfig(serverhttp://localhost:8080)) assertion_urn urn:li:assertion:6e3f9e09-1483-40f9-b9cd-30e5f182694a # Run the assertion assertion_result graph.run_assertion(urnassertion_urn, save_resultTrue) print(fAssertion result (SUCCESS / FAILURE / ERROR): {assertion_result.get(type)})批量与按资产运行的完整写法分别见 run_assertions.pygraph.run_assertions(urns..., save_resultTrue)与 run_assertions_for_asset.pygraph.run_assertions_for_asset(urndataset_urn)支持tag_urns按标签过滤。典型使用场景把断言运行嵌入生产数据管道——例如在批处理作业完成后、下游消费前同步执行数据契约中的关键断言实现管道级熔断Pipeline Circuit Breaking避免坏数据继续向下游传播。新能力二更丰富的 Slack 事件Incident消息v0.3.3 对 Slack 中的事件协作做了三方面增强见 docs/managed-datahub/slack/saas-slack-app.md 的 Manage Data Incidents 一节直接从 Slack 解决Resolve与重新打开Reopen事件同一消息内实时反映事件最新状态由断言触发的事件带有更丰富的细节关联断言信息、影响面等。事件Incident的基本模型在 docs/incidents/incidents.md 中事件被定义为标记数据资产处于不健康状态的独立生命周期概念包含状态active / resolved、标题、描述等字段。两大典型用途健康状态沟通将已知有问题的资产标记为进行中事件消费方在 UI 中可看到健康徽章并跟踪进展管道熔断进阶以事件为基础编排工具可阻塞依赖存在活跃事件输入的数据管道。用 API 自动化事件的创建与解决事件同样可以通过 GraphQL 自动化操作。创建事件使用raiseIncidentmutation返回新事件的 URNmutation raiseIncident { raiseIncident(input: { type: OPERATIONAL title: Dataset Failed Quality Checks description: Dataset failed 2/6 Quality Checks for suite run id xy123mksj812pk23. resourceUrn: urn:li:dataset:(urn:li:dataPlatform:kafka,SampleKafkaDataset,PROD) }) }{ data: { raiseIncident: urn:li:incident:bfecab62-dc10-49a6-a305-78ce0cc6e5b1 } }查询活跃事件则通过实体上的incidents字段可按state如ACTIVE过滤、分页query dataset { dataset(urn: urn:li:dataset:(abc)) { incidents(state: ACTIVE, start: 0, count: 10) { total incidents { urn title description status { state } } } } }与 v0.3.2 的能力衔接值得说明的是事件与断言的联动并非 v0.3.3 首次引入v0.3.2 发布说明docs/managed-datahub/release-notes/v_0_3_2.md中已提到Field assertions 在配置后会于出错时自动拉起事件Group 所有者会在 Slack 事件通知中被标注。v0.3.3 是在此基础上把 Slack 侧的交互与信息密度进一步补强——用户可以在 Slack 消息内直接完成事件的解决与重开无需跳转回 UI。新能力三断言级订阅Assertion-Level Subscriptionsv0.3.3 将订阅粒度细化到了单条断言你可以订阅某条特定断言通过passes、失败fails或出错errors out时的通知。在此之前订阅通常以数据集为粒度粒度细化后可以显著减少通知噪音。订阅模型数据集级 vs 断言级根据 Subscriptions 教程订阅可以在两个层级创建数据集级数据集上的任何变更弃用、Schema 变更、所有者变更等以及该数据集上全部断言的变更都会触发通知断言级仅影响特定断言适合只关心关键质量门禁的场景。与断言相关的变更类型Change Types包括变更类型触发时机ASSERTION_PASSED断言运行通过失败或出错之前会被抑制语义上更接近恢复通知ASSERTION_FAILED断言运行失败ASSERTION_ERROR断言运行报错除此之外订阅还覆盖 Schema 变更OPERATION_COLUMN_ADDED/REMOVED/MODIFIED、操作元数据行插入/更新/删除、事件状态INCIDENT_RAISED/INCIDENT_RESOLVED、弃用DEPRECATED/UNDEPRECATED、摄取状态INGESTION_SUCCEEDED/INGESTION_FAILED等各类事件类型。Python SDK 示例订阅与退订仓库 metadata-ingestion/examples/library/subscription_create.py 给出了完整的订阅代码包括数据集级订阅、按指定变更类型订阅单条断言、以及订阅到群组from datahub.sdk import DataHubClient client DataHubClient.from_env() # 数据集级订阅全部断言变更entity_change_types 默认覆盖数据集全部可用变更类型 client.subscriptions.subscribe( urnurn:li:dataset:(urn:li:dataPlatform:snowflake,purchases,PROD), subscriber_urnurn:li:corpuser:john.doe, ) # 断言级只订阅单条断言的通过/失败 client.subscriptions.subscribe( urnurn:li:assertion:your-assertion-id, subscriber_urnurn:li:corpuser:john.doe, entity_change_types[ASSERTION_PASSED, ASSERTION_FAILED], ) # 群组级订阅失败与出错 client.subscriptions.subscribe( urnurn:li:assertion:your-assertion-id, subscriber_urnurn:li:corpGroup:data-team, entity_change_types[ASSERTION_FAILED, ASSERTION_ERROR], )退订使用subscriptions.unsubscribe支持选择性移除特定变更类型或整体退订示例见 subscription_remove.py。操作前提是调用方具备数据集上的Manage User Subscriptions权限若订阅目标为群组调用方还需是该群组成员。通知的最终触达订阅事件经由 DataHub 的通知体系触达用户根据 docs/managed-datahub/observe/assertions.md 的 Alerts 章节断言告警可通过Slack 私信DM或团队频道、事件告警incident alerts、AWS EventBridge等渠道投递企业级用户还可通过 DataHub Actions 框架 订阅 Kafka 上的 Assertion Change 事件构建自定义响应动作。新能力四结构化属性Structured PropertiesSchema 变更与删除v0.3.3 为结构化属性补齐了 Schema 变更schema change与删除delete能力。结构化属性是附加在数据集、DataJob 等逻辑实体上的带类型、带约束的属性集合是 DataHub 中实现元数据标准化如数据保留期、PII 标记的基础设施完整用法见 Structured Properties 教程。该教程覆盖的能力清单与本版本发布说明的表述相互印证创建结构化属性CLIdatahub properties upsert -f {properties_yaml}或 GraphQLcreateStructuredProperty列出、读取、删除结构化属性为数据集添加结构化属性更新结构化属性并支持破坏性 Schema 变更breaking schema changes——对应 v0.3.3 的 Structured Property Schema Change按结构化属性做搜索与聚合。属性定义示例YAML- id: io.acryl.privacy.retentionTime qualified_name: io.acryl.privacy.retentionTime type: number cardinality: MULTIPLE display_name: Retention Time entity_types: - dataset - dataFlow description: Retention Time is used to figure out how long to retain records in a dataset allowed_values: - value: 30 description: 30 days, usually reserved for datasets that are ephemeral and contain pii - value: 365 description: Use this for non-sensitive data that can be retained for longer从源码结构看OpenAPI 层面向结构化属性暴露了/openapi/v3/entity/structuredProperty端点相关行为在 EntityControllerTest 中可找到覆盖如结构化属性 URN、实体上structuredProperties字段的读写断言可用于验证删除与变更后的读写行为。使用提示string、rich_text、date、urn类型的属性值会被索引为 Elasticsearch / OpenSearch 的 keyword 字段单个值默认上限为 32,766 UTF-8 字节可配置超出时默认被StructuredPropertiesValidator拒绝写入建议短小精炼的值用于可搜索属性大段自由文本应存入实体文档。2.0 UI 修复与其它杂项v0.3.3 同时包含一批 DataHub 2.0 UI 的体验修复新的加载指示器loading indicators修复文本溢出text overflows问题实体健康徽章样式统一consistent entity health badges。这些修复属于新 UI 的打磨性改动与上述数据质量功能共同构成该版本的用户可见变化。上游 OSS DataHub 变更同步v0.3.3 还同步拉取了自 v0.3.2 以来 OSS DataHub 上游仓库的变更对应上游 commit 区间为6ed21bd→92e9a58。这部分内容属于 DataHub Cloud 对开源版本的持续跟随涉及的底层能力如实体注册、GraphQL resolver、OpenAPI 端点等都沉淀在本仓库的各 Java 模块中datahub-graphql-core、metadata-service/openapi-servlet、metadata-io 等感兴趣可进一步深入源码阅读。升级与验证清单综合发布说明与上述各能力文档升级到 v0.3.3 后建议按以下清单验证升级工具链将 CLI/SDK 升级至 v0.13.3覆盖终端、GitHub Actions、Airflow、Python/Java SDK 等所有使用入口。启用并验证按需断言联系 DataHub 团队开启该功能后用runAssertion/runAssertions/runAssertionsForAsset跑通单条、批量与按资产三个入口注意同步运行 30 秒超时与saveResult默认落库行为。验证 Slack 事件闭环确认事件消息可在 Slack 内解决与重开、状态实时刷新并由断言失败触发的告警能携带丰富上下文。收敛通知粒度将高频数据集级订阅收敛为关键断言的断言级订阅并核对ASSERTION_PASSED恢复通知语义、ASSERTION_FAILED、ASSERTION_ERROR三类事件是否按预期投递。演练结构化属性生命周期验证属性 Schema 变更含破坏性变更与删除操作在 CLI / GraphQL / OpenAPI 三条通道上的行为一致。至此v0.3.3 的四大技术主题——按需断言、Slack 事件协作、断言级订阅、结构化属性 Schema 变更与删除——及其 API 与 SDK 用法已全部梳理完毕可直接对照仓库中的教程与示例落地实践。【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考