DataHub SDK v2 Main Client 使用指南统一入口管理实体、搜索与血缘【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahubDataHub 作为数据与 AI 技术栈的上下文平台其 SDK v2 提供了面向对象、类型安全的新一代 Python 客户端。本文以 SDK v2 的DataHubClientMain Client为核心讲解如何通过它统一完成实体管理、搜索、血缘、解析与断言等操作涵盖三种初始化方式、完整配置项、子客户端体系与底层实现原理帮助读者直接在真实 DataHub 实例上落地可运行的集成代码。认识 Main ClientSDK v2 的统一入口在 SDK v2 中DataHubClientmain_client.py被定位为与 DataHub 交互的主客户端它是整个 SDK 的根对象。类注释明确说明了它的职责This class provides the primary interface for interacting with DataHub, including entity management, search, and resolution capabilities.也就是说SDK v2 不再像旧版那样要求开发者直接操作底层的DataHubGraph而是提供一个面向业务的高层门面通过一个DataHubClient实例即可按需获取负责不同领域的子客户端例如实体管理entities、解析resolve、搜索search、血缘lineage、断言assertions和订阅subscriptions。值得注意的是这一代 SDK 仍处于实验阶段。datahub/sdk/__init__.py中的__getattr__会在通过from datahub.sdk import ...导入时发出ExperimentalWarning提示SDK 不适用常规的向后兼容与稳定性保证正式稳定后导入路径将由from datahub.sdk import ...变更为from datahub import ...。因此在使用时建议锁定所用版本并为未来导入路径变更预留适配空间。三种初始化方式与参数校验DataHubClient的构造函数main_client.py通过overload声明了三种互斥的初始化方式方式签名适用场景方式一DataHubClient(*, server: str, token: Optional[str] None)在代码中直接指定 GMS 服务地址与访问令牌最常用方式二DataHubClient(*, config: DatahubClientConfig)需要精细化控制连接参数超时、重试、TLS 等时方式三DataHubClient(*, graph: DataHubGraph)复用已存在的旧版DataHubGraph实例便于平滑迁移方式一server tokenfrom datahub.sdk import DataHubClient client DataHubClient( serverhttp://localhost:8080, tokenyour-access-token, )server即 DataHub GMS 的地址例如http://localhost:8080token是可选的身份令牌。构造时 SDK 会自动把它包装成一个DataHubGraph(configDatahubClientConfig(serverserver, tokentoken))。方式二DatahubClientConfig当需要控制重试、连接池、TLS 证书或请求头时使用DatahubClientConfig。该配置类定义在 config.py完整字段如下字段类型默认值说明serverstr必填DataHub GMS 服务地址tokenOptional[str]None访问令牌与auth二选一authOptional[AuthConfig]None认证提供方配置与token互斥timeout_secOptional[float]None请求超时时间秒retry_status_codesOptional[List[int]]None触发重试的 HTTP 状态码列表retry_max_timesOptional[int]None最大重试次数pool_connectionsOptional[int]None连接池中缓存的最大连接数pool_maxsizeOptional[int]None连接池中单主机最大连接数extra_headersOptional[Dict[str, str]]None附加 HTTP 请求头ca_certificate_pathOptional[str]None自定义 CA 证书路径TLS 校验client_certificate_pathOptional[str]None客户端证书路径双向 TLSclient_key_pathOptional[str]None客户端私钥路径disable_ssl_verificationboolFalse是否关闭 SSL 校验仅建议在可信内网测试环境使用openapi_ingestionOptional[bool]None是否启用 OpenAPI 摄取路径respect_mcp_sync_markerOptional[bool]None是否响应 MCP 中的同步标记marker-aware 同步路由client_modeOptional[ClientMode]None客户端模式INGESTION/CLI/SDKdatahub_componentOptional[str]NoneUser-Agent 中的组件名server_config_refresh_intervalOptional[int]None服务端配置刷新间隔tcp_keepalivebool环境相关是否启用 TCP keepalivedefault_emit_modeOptional[EmitMode]None未显式指定时的默认发射模式from datahub.ingestion.graph.config import DatahubClientConfig from datahub.sdk import DataHubClient config DatahubClientConfig( serverhttps://datahub.example.com, tokentoken, timeout_sec30, retry_status_codes[429, 500, 502, 503, 504], retry_max_times3, pool_maxsize10, extra_headers{X-Custom-Header: value}, ) client DataHubClient(configconfig)需要注意两个约束token与auth不能同时提供_validate_auth_exclusive校验器会抛出ValueError配置对象对未知字段采用extraignore策略多余的键会被静默忽略。方式三复用 DataHubGraphfrom datahub.ingestion.graph.client import DataHubGraph from datahub.sdk import DataHubClient graph DataHubGraph(config...) client DataHubClient(graphgraph)该方式在内部直接把传入的graph保存为self._graph子客户端如EntityClient也正是通过self._client._graph访问底层图能力因此与旧版DataHubGraph混用十分平滑。参数互斥与错误防护构造函数对参数组合做了严格校验main_client.py同时传server与config→ 抛SdkUsageError同时传server与graph→ 抛SdkUsageError同时传config与graph→ 抛SdkUsageError三者皆未提供 → 抛SdkUsageError单元测试 test_client_v2.py 完整覆盖了这些异常分支可作为编写集成代码时的参考。从环境初始化from_env 与 datahub init如果不想把服务地址和令牌硬编码在代码里可以使用类方法from_envmain_client.pyfrom datahub.sdk import DataHubClient client DataHubClient.from_env()其解析顺序如下优先读取环境变量DATAHUB_GMS_URL与DATAHUB_GMS_TOKEN若未设置则读取~/.datahubenv中的凭据——该文件可通过datahub init命令交互式生成。底层调用的是datahub.ingestion.graph.client.get_default_graphclient.py并支持两个仅供内部使用的关键字参数client_mode默认ClientMode.SDK用于区分摄取、CLI 与 SDK 三种调用场景和datahub_component用于在 User-Agent 中标识组件名。ClientMode枚举同样定义在 config.py。连接自检test_connectionclient.test_connection()test_connection()main_client.py是建立连接后第一时间应执行的自检方法它直接委托给底层DataHubGraph.test_connection()用于验证服务可达性、认证令牌是否有效。建议在应用启动阶段调用若失败应捕获异常并给出明确的上游错误信息而不是让后续业务请求逐个失败。子客户端体系一个入口六类能力DataHubClient通过属性按需创建惰性各类子客户端这些子客户端共享同一个_graph底层连接属性返回类型核心能力entitiesEntityClient实体的读取、创建、更新、删除resolveResolverClient按名称/邮箱解析用户、域、术语的 URNsearchSearchClient基于筛选条件的 URN 搜索lineageLineageClient血缘关系的添加与查询、SQL 血缘推断assertionsAssertionsClient自定义断言的同步与结果上报subscriptionsSubscriptionClient订阅管理依赖云端扩展包entities实体管理的核心client.entities返回 EntityClient其构造函数是私有的use DataHubClient.entities instead必须通过主客户端访问。它提供五个关键方法get(urn)按 URN 读取实体支持ContainerUrn、DatasetUrn、DashboardUrn、ChartUrn、GlossaryTermUrn、MLModelUrn等十余种实体类型返回对应的类型化对象如Dataset、ContainerURNs 不存在时抛ItemNotFoundError不支持的实体类型抛SdkUsageErrorcreate(entity)创建实体upsert(entity)存在则更新、不存在则创建update(entity | MetadataPatchProposal)更新实体也可直接提交底层补丁提案delete(urn, check_existsTrue, cascadeFalse, hardFalse)删除实体支持级联删除与硬删除。from datahub.sdk import DataHubClient from datahub.metadata.urns import DatasetUrn client DataHubClient(serverhttp://localhost:8080) # 读取一个数据集并查看其描述 dataset client.entities.get(DatasetUrn(platformhive, namedb.table, envPROD)) print(dataset.description())实体对象本身也是类型化的例如Datasetdataset.py同时提供schema、upstreams、owners、tags、terms、domain、custom_properties等属性与配套的set_*修改方法修改后再通过create/upsert/update提交形成读对象 → 改对象 → 提交的完整闭环。resolve按业务名解析 URNclient.resolve返回ResolverClient。从源码main_client.py可见SDK 会优先尝试导入云服务商扩展acryl_datahub_cloud.sdk.ResolverClient仅在未安装时才回退到开源实现datahub.sdk.resolver_client。其常用方法包括resolve.user(name...)/resolve.user(email...)按用户名或邮箱解析出CorpUserUrnresolve.domain(name...)按名称解析出DomainUrnresolve.term(name...)按名称解析出GlossaryTermUrn。当匹配不到任何实体时抛ItemNotFoundError匹配到多个实体时抛MultipleItemsFoundError这些行为在 test_client_v2.py 中有明确验证。search类型安全的筛选式搜索client.search返回 SearchClient其核心方法是get_urns(queryNone, filterNone)返回匹配实体的 URN 迭代器。结合FilterDslsearch_filters.py可以声明式地组合筛选条件from datahub.sdk import DataHubClient from datahub.sdk.search_filters import FilterDsl client DataHubClient.from_env() # 查找所有属于 snowflake 平台、且在指定域下的数据集 URN f FilterDsl.and_( FilterDsl.entity_type(dataset), FilterDsl.platform(snowflake), FilterDsl.domain(urn:li:domain:analytics), ) urns client.search.get_urns(query*, filterf) for urn in urns: print(urn)FilterDsl提供的筛选原语包括entity_type、entity_subtype、platform、domain、container支持direct_descendants_only、env、owner、glossary_term、tag、has_custom_property、custom_filter、soft_deleted并支持and_/or_/not_组合。值得注意的是 search_client.py 中的compile_filters即使未显式指定状态筛选SDK 也会自动附加soft_deleted(NOT_SOFT_DELETED)条件默认排除软删除实体避免脏数据混入搜索结果。lineage血缘的增查与推断client.lineage返回 LineageClient提供了丰富的血缘操作add_lineage(upstream..., downstream..., column_lineage..., transformation_text...)为数据集、数据任务、仪表盘、图表之间建立血缘column_lineage可传布尔值、字段映射或auto_fuzzy/auto_strict让 SDK 自动推断列级血缘infer_lineage_from_sql(query_text..., platform..., ...)基于 SQL 文本经由 sqlglot 解析自动推断血缘get_lineage(source_urn..., directionupstream|downstream, max_hops1, count500, ...)查询血缘结果add_dataset_copy_lineage/add_dataset_transform_lineage/add_datajob_lineage面向常见场景的快捷方法。client.lineage.add_lineage( upstreamurn:li:dataset:(urn:li:dataPlatform:hive,raw_events,PROD), downstreamurn:li:dataset:(urn:li:dataPlatform:hive,agg_events,PROD), column_lineageauto_strict, )assertions 与 subscriptions质量与订阅assertions属性会优先尝试加载云端扩展acryl_datahub_cloud.sdk.AssertionsClient若云端包缺失则回退到开源的 assertion_client.py其核心方法为sync_custom_assertion(...)同步自定义断言与report_assertion_result(...)上报断言执行结果类型为SUCCESS/FAILURE/ERROR/INIT。subscriptions属性仅能通过云端扩展包acryl-datahub-cloud提供若未安装会抛出带安装提示的SdkUsageErrorplease install it withpip install acryl-datahub-cloud。可见 SDK 采用了开源核心 云端可选能力的优雅降级策略。完整实战示例把上述能力串起来一个典型的数据集治理脚本如下from datahub.ingestion.graph.config import DatahubClientConfig from datahub.sdk import DataHubClient from datahub.sdk.search_filters import FilterDsl # 1. 初始化并自检连接 client DataHubClient( configDatahubClientConfig( serverhttps://datahub.example.com, tokentoken, timeout_sec30, retry_max_times3, ) ) client.test_connection() # 2. 按业务名解析用户与域 owner_urn client.resolve.user(emailaliceexample.com) domain_urn client.resolve.domain(nameanalytics) # 3. 搜索某平台下指定域的数据集 filter FilterDsl.and_( FilterDsl.entity_type(dataset), FilterDsl.platform(snowflake), FilterDsl.domain(domain_urn), ) for urn in client.search.get_urns(filterfilter): dataset client.entities.get(urn) if dataset.owners() is None: dataset.set_owners([owner_urn]) client.entities.update(dataset) print(f已为 {urn} 补充 owner)从源码与测试理解实现细节统一的底层委托虽然DataHubClient对外暴露的是高层 API但它本质上是旧版DataHubGraphclient.py约 2400 行的薄封装_graph承担了所有 GraphQL/OpenAPI 请求、重试、连接池与遥测。EntityClient._graph直接透传self._client._graph子客户端之间共享同一个连接避免重复建连。惰性子客户端与云端降级resolve、assertions、subscriptions三个属性都采用先尝试acryl_datahub_cloud.sdk失败则回退或报错的模式且对回退路径上的ImportError做了精细化区分main_client.py只有明确由云端包缺失引起的ImportError才回退其他异常如云端包内部依赖损坏会被原样重新抛出避免掩盖真实故障。测试印证SDK v2 的测试集中在 metadata-ingestion/tests/unit/sdk_v2/test_client_v2.py覆盖客户端三种创建方式、非法参数组合抛错、resolve.user的命中/缺失/多义场景test_entity_client.py验证实体 get/create/upsert/update/delete 行为test_search_client.py验证筛选编译与默认软删除过滤test_lineage_client.py验证血缘添加与 SQL 推断test_assertions_client.py验证断言同步与结果上报。这些测试大多基于Mock(specDataHubGraph)的桩对象编写说明子客户端的核心逻辑可以脱离真实服务进行单元验证也为读者在 CI 中模拟 SDK 行为提供了范式。相关文档索引SDK v2 的 Sphinx API 文档均位于 docs-website/sphinx/apidocs/sdk-v2/除本文对应的 main-client.rst 外还包括 entity-client.rst、search-client.rst、lineage-client.rst、resolver-client.rst 与 entities.rst。它们均由 Sphinxautomodule指令自动生成源码即文档——metadata-ingestion/src/datahub/sdk/ 目录下的 docstring 与类型注解是理解各客户端最权威的参考资料。【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 DataHub SDK v2 Main Client 使用指南：统一入口管理实体、搜索与血缘
DataHub SDK v2 Main Client 使用指南统一入口管理实体、搜索与血缘【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahubDataHub 作为数据与 AI 技术栈的上下文平台其 SDK v2 提供了面向对象、类型安全的新一代 Python 客户端。本文以 SDK v2 的DataHubClientMain Client为核心讲解如何通过它统一完成实体管理、搜索、血缘、解析与断言等操作涵盖三种初始化方式、完整配置项、子客户端体系与底层实现原理帮助读者直接在真实 DataHub 实例上落地可运行的集成代码。认识 Main ClientSDK v2 的统一入口在 SDK v2 中DataHubClientmain_client.py被定位为与 DataHub 交互的主客户端它是整个 SDK 的根对象。类注释明确说明了它的职责This class provides the primary interface for interacting with DataHub, including entity management, search, and resolution capabilities.也就是说SDK v2 不再像旧版那样要求开发者直接操作底层的DataHubGraph而是提供一个面向业务的高层门面通过一个DataHubClient实例即可按需获取负责不同领域的子客户端例如实体管理entities、解析resolve、搜索search、血缘lineage、断言assertions和订阅subscriptions。值得注意的是这一代 SDK 仍处于实验阶段。datahub/sdk/__init__.py中的__getattr__会在通过from datahub.sdk import ...导入时发出ExperimentalWarning提示SDK 不适用常规的向后兼容与稳定性保证正式稳定后导入路径将由from datahub.sdk import ...变更为from datahub import ...。因此在使用时建议锁定所用版本并为未来导入路径变更预留适配空间。三种初始化方式与参数校验DataHubClient的构造函数main_client.py通过overload声明了三种互斥的初始化方式方式签名适用场景方式一DataHubClient(*, server: str, token: Optional[str] None)在代码中直接指定 GMS 服务地址与访问令牌最常用方式二DataHubClient(*, config: DatahubClientConfig)需要精细化控制连接参数超时、重试、TLS 等时方式三DataHubClient(*, graph: DataHubGraph)复用已存在的旧版DataHubGraph实例便于平滑迁移方式一server tokenfrom datahub.sdk import DataHubClient client DataHubClient( serverhttp://localhost:8080, tokenyour-access-token, )server即 DataHub GMS 的地址例如http://localhost:8080token是可选的身份令牌。构造时 SDK 会自动把它包装成一个DataHubGraph(configDatahubClientConfig(serverserver, tokentoken))。方式二DatahubClientConfig当需要控制重试、连接池、TLS 证书或请求头时使用DatahubClientConfig。该配置类定义在 config.py完整字段如下字段类型默认值说明serverstr必填DataHub GMS 服务地址tokenOptional[str]None访问令牌与auth二选一authOptional[AuthConfig]None认证提供方配置与token互斥timeout_secOptional[float]None请求超时时间秒retry_status_codesOptional[List[int]]None触发重试的 HTTP 状态码列表retry_max_timesOptional[int]None最大重试次数pool_connectionsOptional[int]None连接池中缓存的最大连接数pool_maxsizeOptional[int]None连接池中单主机最大连接数extra_headersOptional[Dict[str, str]]None附加 HTTP 请求头ca_certificate_pathOptional[str]None自定义 CA 证书路径TLS 校验client_certificate_pathOptional[str]None客户端证书路径双向 TLSclient_key_pathOptional[str]None客户端私钥路径disable_ssl_verificationboolFalse是否关闭 SSL 校验仅建议在可信内网测试环境使用openapi_ingestionOptional[bool]None是否启用 OpenAPI 摄取路径respect_mcp_sync_markerOptional[bool]None是否响应 MCP 中的同步标记marker-aware 同步路由client_modeOptional[ClientMode]None客户端模式INGESTION/CLI/SDKdatahub_componentOptional[str]NoneUser-Agent 中的组件名server_config_refresh_intervalOptional[int]None服务端配置刷新间隔tcp_keepalivebool环境相关是否启用 TCP keepalivedefault_emit_modeOptional[EmitMode]None未显式指定时的默认发射模式from datahub.ingestion.graph.config import DatahubClientConfig from datahub.sdk import DataHubClient config DatahubClientConfig( serverhttps://datahub.example.com, tokentoken, timeout_sec30, retry_status_codes[429, 500, 502, 503, 504], retry_max_times3, pool_maxsize10, extra_headers{X-Custom-Header: value}, ) client DataHubClient(configconfig)需要注意两个约束token与auth不能同时提供_validate_auth_exclusive校验器会抛出ValueError配置对象对未知字段采用extraignore策略多余的键会被静默忽略。方式三复用 DataHubGraphfrom datahub.ingestion.graph.client import DataHubGraph from datahub.sdk import DataHubClient graph DataHubGraph(config...) client DataHubClient(graphgraph)该方式在内部直接把传入的graph保存为self._graph子客户端如EntityClient也正是通过self._client._graph访问底层图能力因此与旧版DataHubGraph混用十分平滑。参数互斥与错误防护构造函数对参数组合做了严格校验main_client.py同时传server与config→ 抛SdkUsageError同时传server与graph→ 抛SdkUsageError同时传config与graph→ 抛SdkUsageError三者皆未提供 → 抛SdkUsageError单元测试 test_client_v2.py 完整覆盖了这些异常分支可作为编写集成代码时的参考。从环境初始化from_env 与 datahub init如果不想把服务地址和令牌硬编码在代码里可以使用类方法from_envmain_client.pyfrom datahub.sdk import DataHubClient client DataHubClient.from_env()其解析顺序如下优先读取环境变量DATAHUB_GMS_URL与DATAHUB_GMS_TOKEN若未设置则读取~/.datahubenv中的凭据——该文件可通过datahub init命令交互式生成。底层调用的是datahub.ingestion.graph.client.get_default_graphclient.py并支持两个仅供内部使用的关键字参数client_mode默认ClientMode.SDK用于区分摄取、CLI 与 SDK 三种调用场景和datahub_component用于在 User-Agent 中标识组件名。ClientMode枚举同样定义在 config.py。连接自检test_connectionclient.test_connection()test_connection()main_client.py是建立连接后第一时间应执行的自检方法它直接委托给底层DataHubGraph.test_connection()用于验证服务可达性、认证令牌是否有效。建议在应用启动阶段调用若失败应捕获异常并给出明确的上游错误信息而不是让后续业务请求逐个失败。子客户端体系一个入口六类能力DataHubClient通过属性按需创建惰性各类子客户端这些子客户端共享同一个_graph底层连接属性返回类型核心能力entitiesEntityClient实体的读取、创建、更新、删除resolveResolverClient按名称/邮箱解析用户、域、术语的 URNsearchSearchClient基于筛选条件的 URN 搜索lineageLineageClient血缘关系的添加与查询、SQL 血缘推断assertionsAssertionsClient自定义断言的同步与结果上报subscriptionsSubscriptionClient订阅管理依赖云端扩展包entities实体管理的核心client.entities返回 EntityClient其构造函数是私有的use DataHubClient.entities instead必须通过主客户端访问。它提供五个关键方法get(urn)按 URN 读取实体支持ContainerUrn、DatasetUrn、DashboardUrn、ChartUrn、GlossaryTermUrn、MLModelUrn等十余种实体类型返回对应的类型化对象如Dataset、ContainerURNs 不存在时抛ItemNotFoundError不支持的实体类型抛SdkUsageErrorcreate(entity)创建实体upsert(entity)存在则更新、不存在则创建update(entity | MetadataPatchProposal)更新实体也可直接提交底层补丁提案delete(urn, check_existsTrue, cascadeFalse, hardFalse)删除实体支持级联删除与硬删除。from datahub.sdk import DataHubClient from datahub.metadata.urns import DatasetUrn client DataHubClient(serverhttp://localhost:8080) # 读取一个数据集并查看其描述 dataset client.entities.get(DatasetUrn(platformhive, namedb.table, envPROD)) print(dataset.description())实体对象本身也是类型化的例如Datasetdataset.py同时提供schema、upstreams、owners、tags、terms、domain、custom_properties等属性与配套的set_*修改方法修改后再通过create/upsert/update提交形成读对象 → 改对象 → 提交的完整闭环。resolve按业务名解析 URNclient.resolve返回ResolverClient。从源码main_client.py可见SDK 会优先尝试导入云服务商扩展acryl_datahub_cloud.sdk.ResolverClient仅在未安装时才回退到开源实现datahub.sdk.resolver_client。其常用方法包括resolve.user(name...)/resolve.user(email...)按用户名或邮箱解析出CorpUserUrnresolve.domain(name...)按名称解析出DomainUrnresolve.term(name...)按名称解析出GlossaryTermUrn。当匹配不到任何实体时抛ItemNotFoundError匹配到多个实体时抛MultipleItemsFoundError这些行为在 test_client_v2.py 中有明确验证。search类型安全的筛选式搜索client.search返回 SearchClient其核心方法是get_urns(queryNone, filterNone)返回匹配实体的 URN 迭代器。结合FilterDslsearch_filters.py可以声明式地组合筛选条件from datahub.sdk import DataHubClient from datahub.sdk.search_filters import FilterDsl client DataHubClient.from_env() # 查找所有属于 snowflake 平台、且在指定域下的数据集 URN f FilterDsl.and_( FilterDsl.entity_type(dataset), FilterDsl.platform(snowflake), FilterDsl.domain(urn:li:domain:analytics), ) urns client.search.get_urns(query*, filterf) for urn in urns: print(urn)FilterDsl提供的筛选原语包括entity_type、entity_subtype、platform、domain、container支持direct_descendants_only、env、owner、glossary_term、tag、has_custom_property、custom_filter、soft_deleted并支持and_/or_/not_组合。值得注意的是 search_client.py 中的compile_filters即使未显式指定状态筛选SDK 也会自动附加soft_deleted(NOT_SOFT_DELETED)条件默认排除软删除实体避免脏数据混入搜索结果。lineage血缘的增查与推断client.lineage返回 LineageClient提供了丰富的血缘操作add_lineage(upstream..., downstream..., column_lineage..., transformation_text...)为数据集、数据任务、仪表盘、图表之间建立血缘column_lineage可传布尔值、字段映射或auto_fuzzy/auto_strict让 SDK 自动推断列级血缘infer_lineage_from_sql(query_text..., platform..., ...)基于 SQL 文本经由 sqlglot 解析自动推断血缘get_lineage(source_urn..., directionupstream|downstream, max_hops1, count500, ...)查询血缘结果add_dataset_copy_lineage/add_dataset_transform_lineage/add_datajob_lineage面向常见场景的快捷方法。client.lineage.add_lineage( upstreamurn:li:dataset:(urn:li:dataPlatform:hive,raw_events,PROD), downstreamurn:li:dataset:(urn:li:dataPlatform:hive,agg_events,PROD), column_lineageauto_strict, )assertions 与 subscriptions质量与订阅assertions属性会优先尝试加载云端扩展acryl_datahub_cloud.sdk.AssertionsClient若云端包缺失则回退到开源的 assertion_client.py其核心方法为sync_custom_assertion(...)同步自定义断言与report_assertion_result(...)上报断言执行结果类型为SUCCESS/FAILURE/ERROR/INIT。subscriptions属性仅能通过云端扩展包acryl-datahub-cloud提供若未安装会抛出带安装提示的SdkUsageErrorplease install it withpip install acryl-datahub-cloud。可见 SDK 采用了开源核心 云端可选能力的优雅降级策略。完整实战示例把上述能力串起来一个典型的数据集治理脚本如下from datahub.ingestion.graph.config import DatahubClientConfig from datahub.sdk import DataHubClient from datahub.sdk.search_filters import FilterDsl # 1. 初始化并自检连接 client DataHubClient( configDatahubClientConfig( serverhttps://datahub.example.com, tokentoken, timeout_sec30, retry_max_times3, ) ) client.test_connection() # 2. 按业务名解析用户与域 owner_urn client.resolve.user(emailaliceexample.com) domain_urn client.resolve.domain(nameanalytics) # 3. 搜索某平台下指定域的数据集 filter FilterDsl.and_( FilterDsl.entity_type(dataset), FilterDsl.platform(snowflake), FilterDsl.domain(domain_urn), ) for urn in client.search.get_urns(filterfilter): dataset client.entities.get(urn) if dataset.owners() is None: dataset.set_owners([owner_urn]) client.entities.update(dataset) print(f已为 {urn} 补充 owner)从源码与测试理解实现细节统一的底层委托虽然DataHubClient对外暴露的是高层 API但它本质上是旧版DataHubGraphclient.py约 2400 行的薄封装_graph承担了所有 GraphQL/OpenAPI 请求、重试、连接池与遥测。EntityClient._graph直接透传self._client._graph子客户端之间共享同一个连接避免重复建连。惰性子客户端与云端降级resolve、assertions、subscriptions三个属性都采用先尝试acryl_datahub_cloud.sdk失败则回退或报错的模式且对回退路径上的ImportError做了精细化区分main_client.py只有明确由云端包缺失引起的ImportError才回退其他异常如云端包内部依赖损坏会被原样重新抛出避免掩盖真实故障。测试印证SDK v2 的测试集中在 metadata-ingestion/tests/unit/sdk_v2/test_client_v2.py覆盖客户端三种创建方式、非法参数组合抛错、resolve.user的命中/缺失/多义场景test_entity_client.py验证实体 get/create/upsert/update/delete 行为test_search_client.py验证筛选编译与默认软删除过滤test_lineage_client.py验证血缘添加与 SQL 推断test_assertions_client.py验证断言同步与结果上报。这些测试大多基于Mock(specDataHubGraph)的桩对象编写说明子客户端的核心逻辑可以脱离真实服务进行单元验证也为读者在 CI 中模拟 SDK 行为提供了范式。相关文档索引SDK v2 的 Sphinx API 文档均位于 docs-website/sphinx/apidocs/sdk-v2/除本文对应的 main-client.rst 外还包括 entity-client.rst、search-client.rst、lineage-client.rst、resolver-client.rst 与 entities.rst。它们均由 Sphinxautomodule指令自动生成源码即文档——metadata-ingestion/src/datahub/sdk/ 目录下的 docstring 与类型注解是理解各客户端最权威的参考资料。【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考