DataHub Python SDK Search Client 实战：用代码搜索、过滤与发现数据资产

DataHub Python SDK Search Client 实战用代码搜索、过滤与发现数据资产【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahubDataHub 的 Python SDKacryl-datahub提供了开箱即用的 Search Client让开发者可以用少量 Python 代码对数据集、图表、仪表盘等元数据实体进行关键词检索、结构化过滤与逻辑组合查询。本文基于仓库中的官方教程与源码实现系统讲解如何安装连接、使用 Query 与 Filter 两种搜索方式、组合AND/OR/NOT逻辑并深入 FilterDsl 底层原理帮助你快速构建属于自己的元数据发现工具。环境准备安装 SDK 并连接 DataHub 实例使用 Search SDK 前需要先安装acryl-datahub包并建立与 DataHub 实例的连接。安装与 CLI 使用的完整说明可参考仓库中的 CLI 使用指南 与 DataHub CLI 文档。连接 DataHub 实例只需一行代码见 search_with_query.pyfrom datahub.sdk import DataHubClient client DataHubClient(serveryour_server, tokenyour_token)两个关键参数serverDataHub GMS 服务的 URL。本地部署http://localhost:8080托管部署https://your_datahub_url/gmstoken需要在 DataHub 实例中生成 Personal Access Token 后传入生成方式见 Personal Access Token 指南。除显式传入连接参数外SDK 还支持从环境变量读取配置例如 search_filter_by_domain.py 中使用的DataHubClient.from_env()适合在 CI/CD、脚本化任务中避免硬编码敏感信息。两种搜索方式DataHub 的 Search SDK 提供两种互补的搜索方式二者可独立使用也可组合使用以获得更精确的结果搜索方式原理适用场景Query-based search关键词搜索用简单关键字匹配名称、描述、列名等常见字段探索性搜索不确定目标资产的具体信息时Filter-based search过滤搜索用结构化过滤器按平台、环境、实体类型等字段圈定结果需要精确缩小到特定资产类型或元数据字段时Query 与 Filter 可以同时使用例如查找所有与 forecast 相关的 Snowflake 数据集这正是下文组合 Query 与 Filter要演示的场景。关键词搜索查找所有与 sales 相关的实体client.search.get_urns(querysales)会检索元数据中包含sales的所有资产名称、描述、列名等字段返回一个可迭代的 URN 列表from datahub.sdk import DataHubClient client DataHubClient(serveryour_server, tokenyour_token) # 检索元数据中包含 sales 的实体 results client.search.get_urns(querysales) print(list(results))示例输出URN 格式[ DatasetUrn(urn:li:dataset:(urn:li:dataPlatform:snowflake,sales_revenue_2023,PROD)), DatasetUrn(urn:li:dataset:(urn:li:dataPlatform:snowflake,sales_forecast,PROD)) ]过滤搜索查找所有 Snowflake 实体按平台过滤使用F.platform(snowflake)会将platform规范化为平台 URN 后匹配见 search_with_filter.pyfrom datahub.sdk import DataHubClient, FilterDsl as F client DataHubClient(serveryour_server, tokenyour_token) # 检索所有位于 snowflake 平台上的实体 results client.search.get_urns(filterF.platform(snowflake)) print(list(results))组合 Query 与 Filter查找与 forecast 相关的 Snowflake 数据集get_urns()的query与filter参数可以同时传入实现内容 结构双重约束见 search_with_query_and_filter.pyfrom datahub.sdk import DataHubClient, FilterDsl as F client DataHubClient(serveryour_server, tokenyour_token) # 检索元数据包含 forecast 且同时满足平台snowflake、类型dataset 的实体 results client.search.get_urns( queryforecast, filterF.and_(F.platform(snowflake), F.entity_type(dataset)) ) print(list(results))这里用F.and_()将两个过滤条件组合实现要么是 chart要么是 Snowflake dataset之类的复杂语义。常用搜索模式仓库的 examples/library 目录 提供了一整套开箱即用的搜索示例覆盖了最常见的过滤场景。查找所有 Dashboard按实体类型过滤entity_type支持dataset、chart、dashboard、corpuser、dataProduct等类型见 search_filter_by_entity_type.pyfrom datahub.sdk import DataHubClient from datahub.sdk.search_filters import FilterDsl as F client DataHubClient(serveryour_server, tokenyour_token) results client.search.get_urns(filterF.entity_type(dashboard))查找所有 Snowflake 平台实体按平台过滤见 search_filter_by_platform.pyfrom datahub.sdk import DataHubClient from datahub.sdk.search_filters import FilterDsl as F client DataHubClient(serveryour_server, tokenyour_token) results client.search.get_urns(filterF.platform(snowflake))查找生产环境中的所有实体按环境过滤使用F.env(PROD)见 search_filter_by_env.pyfrom datahub.sdk import DataHubClient from datahub.sdk.search_filters import FilterDsl as F client DataHubClient(serveryour_server, tokenyour_token) results client.search.get_urns(filterF.env(PROD))注意并非所有实体类型都有env概念例如 dashboard、chart。源码_EnvFilter的实现注释也明确指出如果指定了 env 过滤这些没有环境属性的实体类型会被排除在结果之外。详见下文环境过滤的双字段实现。查找特定 Domain 中的所有实体按域过滤需要传入 Domain 的完整 URN见 search_filter_by_domain.pyfrom datahub.sdk import DataHubClient from datahub.sdk.search_filters import FilterDsl as F # 查找 marketing 域中的所有资产支持 from_env 从环境变量读取连接配置 client DataHubClient.from_env() results client.search.get_urns(filterF.domain(urn:li:domain:marketing))查找特定子类型的所有实体子类型Subtype用于表达同一平台内部的细分概念如 MLflow 平台下的ML Experiment见 search_filter_by_entity_subtype.pyfrom datahub.sdk import DataHubClient from datahub.sdk.search_filters import FilterDsl as F # 查找 mlflow 平台下所有子类型为 ML Experiment 的资产 client DataHubClient(serveryour_server, tokenyour_token) results client.search.get_urns( filterF.and_(F.platform(mlflow), F.entity_subtype(ML Experiment)) )查找具有特定自定义属性的所有实体F.has_custom_property(key, value)用于匹配资产的自定义属性见 search_filter_by_custom_property.pyfrom datahub.sdk import DataHubClient from datahub.sdk.search_filters import FilterDsl as F client DataHubClient(serveryour_server, tokenyour_token) # 查找自定义属性 my_custom_property 等于 my_value 的所有资产 results client.search.get_urns( filterF.has_custom_property(my_custom_property, my_value) )从源码看has_custom_property实际是custom_filter(fieldcustomProperties, conditionEQUAL, values[keyvalue])的便捷封装见 search_filters.py。逻辑组合查找所有 Charts 与 Snowflake 数据集使用F.and_、F.or_、F.not_可以自由组合过滤条件构建任意复杂的布尔表达式见 search_filter_combined_operation.pyfrom datahub.sdk import DataHubClient, FilterDsl as F client DataHubClient(serveryour_server, tokenyour_token) # 查找所有 chart或者所有 snowflake 数据集 results client.search.get_urns( filterF.or_( F.entity_type(chart), F.and_(F.platform(snowflake), F.entity_type(dataset)), ) ) print(list(results))逻辑取反查找不在生产环境中的所有 ChartsF.not_()用于排除满足某条件的实体见 search_filter_not.pyfrom datahub.sdk import DataHubClient, FilterDsl as F client DataHubClient(serveryour_server, tokenyour_token) # 查找所有 chart且不在 PROD 环境中的实体 results client.search.get_urns( filterF.and_(F.entity_type(chart), F.not_(F.env(PROD))), ) print(list(results))高级按其他可搜索字段自定义过滤当内置过滤器不够用时可以使用F.custom_filter()直接指定字段、条件与取值见 search_filter_custom.pyfrom datahub.sdk import DataHubClient, FilterDsl as F client DataHubClient(serveryour_server, tokenyour_token) # 查找 urn 中包含 example_dataset 的数据集 results client.search.get_urns( filterF.custom_filter(fieldurn, conditionCONTAIN, values[example_dataset]) ) print(list(results))F.custom_filter()可以作用的字段是 PDL 模型中标注了Searchable注解的字段。例如DataJobInfo.pdl中的name、description、env字段均被标注为可搜索字段因此可以对 datajob 实体按这些字段过滤详见 DataJobInfo.pdl。FilterDsl 完整参考FilterDsl通常以F为别名导入是构造所有过滤器的唯一入口定义于 search_filters.py。除本文已演示的过滤器外它还包括owner、glossary_term、tag、container、soft_deleted等能力。过滤器选项一览过滤器类型示例代码底层匹配字段平台 PlatformF.platform(snowflake)platform.keyword环境 EnvironmentF.env(PROD)origin/env实体类型 Entity TypeF.entity_type(dataset)_entityType域 DomainF.domain(urn:li:domain:xyz)domains子类型 SubtypeF.entity_subtype(ML Experiment)typeNames删除状态 Deletion StatusF.soft_deleted(NOT_SOFT_DELETED)状态过滤规则自定义属性 Custom PropertyF.has_custom_property(department, sales)customProperties容器 ContainerF.container(urn:li:container:xyz, direct_descendants_onlyTrue)container/browsePathV2所有者 OwnerF.owner(urn:li:corpuser:jdoe)owners术语 Glossary TermF.glossary_term(urn:li:glossaryTerm:xyz)glossaryTerms标签 TagF.tag(urn:li:tag:xyz)tags其中平台、域、容器、所有者、术语等过滤器在构造时会自动进行 URN 校验与规范化例如平台会被转换为DataPlatformUrn域会被转换为DomainUrn传入非法 URN 会直接抛出校验错误避免把错误查询发到服务端。container过滤还支持direct_descendants_only参数为True时只返回容器的直接后代。逻辑运算符选项运算符示例代码含义ANDF.and_(F.platform(snowflake), F.entity_type(dataset))返回满足所有条件的实体ORF.or_(F.entity_type(chart), F.entity_type(dataset))返回满足至少一个条件的实体NOTF.not_(F.env(PROD))排除满足给定条件的实体此外FilterDsl还提供F.true()与F.false()两个恒真/恒假过滤器分别编译为urn EXISTS及其取反在需要以程序化方式动态拼接过滤表达式时非常有用。custom_filter 支持的匹配条件F.custom_filter(field, condition, values)的condition参数取值与底层 GraphQLFilterOperator枚举对齐完整枚举定义见 filters.py。文档明确列出的条件如下条件说明EQUAL字符串字段精确匹配CONTAIN字符串字段包含指定子串START_WITH以指定子串开头END_WITH以指定子串结尾GREATER_THAN数值或时间戳字段大于指定值LESS_THAN数值或时间戳字段小于指定值除了上述 6 种底层FilterOperator还支持IEQUAL忽略大小写的精确匹配、IN、EXISTS、GREATER_THAN_OR_EQUAL_TO、LESS_THAN_OR_EQUAL_TO、DESCENDANTS_INCL、ANCESTORS_INCL、RELATED_INCL等条件可用于容器层级、血缘关系等更复杂的场景。深入底层get_urns 的执行链路理解 SDK 的执行链路有助于排查问题与预测行为。核心入口是SearchClient.get_urns()定义于 search_client.pydef get_urns( self, query: Optional[str] None, filter: Optional[Filter] None, *, skip_cache: bool False, ) - Iterable[Urn]: types, compiled_filters compile_filters(filter) for urn in self._client._graph.get_urns_by_filter( queryquery, statusNone, extra_or_filterscompiled_filters, entity_typestypes, skip_cacheskip_cache, ): yield Urn.from_string(urn)整个流程可拆解为三步编译过滤器compile_filters()遍历过滤表达式的语法树每个过滤器都有dfs()与compile()方法将其转换为底层的_OrFilters结构即[{and: [SearchFilterRule, ...]}, ...]每个SearchFilterRule由field、condition、values三个字段组成见 filters.py。推断实体类型集合compute_entity_types()从编译结果中提取所有entity_type过滤条件作为搜索的实体类型范围传给服务端。如果没有指定任何实体类型过滤则使用与 DataHub UI 一致的默认搜索集合见 search_client.py。调用底层 Graph 客户端将 query、过滤规则与实体类型一并交给_graph.get_urns_by_filter()执行真实搜索并把返回的 URN 字符串包装为Urn对象逐个产出。值得注意的是get_urns返回的是一个惰性的Iterable[Urn]配合list()或循环使用即可消费所有结果。默认过滤自动排除已删除实体compile_filters()还有一个隐式行为如果用户没有显式指定删除状态过滤SDK 会自动追加soft_deleted(RemovedStatusFilter.NOT_SOFT_DELETED)即默认只返回未被软删除的实体见 search_client.py。RemovedStatusFilter枚举支持三档取值见 filters.pyNOT_SOFT_DELETED仅搜索未标记为删除的实体默认ALL搜索所有实体包含已删除ONLY_SOFT_DELETED仅搜索已软删除的实体环境过滤的双字段实现F.env()的底层实现比较特殊见 search_filters.py它编译为两个 OR 分支——大多数实体类型匹配origin字段而容器类实体匹配env字段。也就是说F.env(PROD)实际等价于origin PROD或env PROD从而同时覆盖普通资产与容器资产两种存储方式。布尔代数与取反的编译细节逻辑组合并非简单的字符串拼接。F.and_/F.or_/F.not_构造出_And/_Or/_Not过滤器树_Not会对其内部过滤器的规则调用negate()方法将SearchFilterRule.negated置反。因此在search_filter_not.py的示例中F.not_(F.env(PROD))最终会生成带negatedtrue标记的过滤规则见 filters.py并由服务端解释为排除语义。FAQ如何完成身份认证在 DataHub 实例的设置中生成 Personal Access Token通过DataHubClient(server..., token...)传入即可详细步骤见 Personal Access Token 指南。SDK 也支持DataHubClient.from_env()从环境变量读取连接配置。能否同时使用 Query 和 Filter可以。get_urns(query..., filter...)两个参数可同时传入实现关键词 结构化条件的联合检索这正是查找所有与 forecast 相关的 Snowflake 数据集示例所演示的用法。返回结果是什么类型如何消费get_urns()返回惰性的Iterable[Urn]每个元素是Urn对象。直接list(results)即可得到 URN 列表也可以遍历并基于 URN 继续调用 SDK 的实体读取能力做进一步分析。更多示例仓库 metadata-ingestion/examples/library 还提供了大量可运行的搜索示例包括按标签过滤的 search_filter_by_tag.py、按所有者过滤的 search_filter_by_owner.py、按术语过滤的 search_filter_by_glossary_term.py、独立的 search_filter_and.py 与 search_filter_or.py以及展示全部过滤能力的 search_filter_options.py。结合本文讲解的 FilterDsl 参考表你可以快速将这些模式复用到自己的数据发现、数据治理巡检或自动化运维脚本中。【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考