DataHub 数据集自定义属性Custom Properties实战指南通过 Python SDK 与 Java 实现添加、移除与替换【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub自定义属性是 DataHub 中一种轻量、灵活的元数据扩展方式它允许你在 Dataset 等实体上维护任意数量的字符串键值对Key-Value Map用于补充标准元数据字段无法表达的业务上下文例如数据单位、覆盖的时间范围、所属地理区域等。本指南以数据集fct_users_deleted为例完整演示如何在 DataHub 中通过 Python SDK 与 Java SDK 对自定义属性执行添加Add添加与移除Add Remove整体替换Replace三种操作并结合仓库源码与元数据模型说明其底层实现原理帮助你在实际项目中安全、精准地维护自定义属性同时利用它们增强搜索与发现能力。为什么要在数据集上使用自定义属性DataHub 的标准元数据字段如name、description覆盖面有限而真实业务中数据往往带有大量只有团队内部才懂的附加信息。自定义属性正是为这类场景设计的补充业务上下文描述数据的特定属性例如使用的计量单位units、覆盖的日期范围date_range、数据所属的地理区域region等。在大型复杂数据集上这些额外上下文能确保数据被正确、有效地使用。支撑高级搜索与发现通过将属性写入元数据索引用户可以基于特定属性对数据集进行过滤与排序快速定位所需数据而无需人工翻阅大量数据集。从元数据模型上看DataHub 将 Dataset 的自定义属性建模为字符串键值对的映射map of key-value pairs of strings。该模型定义于 CustomProperties.pdl其核心字段为record CustomProperties { customProperties: map[string, string] { } }值得注意的是该字段带有Searchable注解且配置了/*通配路径、fieldType: TEXT、queryByDefault: true意味着所有自定义属性默认参与全文检索这正是自定义属性可以支撑高级搜索与过滤的模型层依据。本指南目标本指南将围绕数据集fct_users_deleted演示三种操作Add添加向数据集追加自定义属性不影响已有属性Remove移除从数据集删除指定属性不影响其他属性Replace替换整体替换整个属性 Map不影响同一 Aspect 中的其他字段例如DatasetPropertiesAspect 中与customProperties同级的name、description等字段。底层模型datasetPropertiesAspect 与customPropertiesMap在 DataHub 中自定义属性并不是孤立存在的它们存放在 Dataset 的datasetPropertiesAspect 中。该 Aspect 的定义位于 DatasetProperties.pdl其中record DatasetProperties includes CustomProperties, ExternalReference即它继承了CustomProperties的customProperties: map[string, string]字段同时自身还包含name、qualifiedName、description、created、lastModified等字段。这一点解释了为什么三种操作之间存在本质区别对customProperties这个 Map 内部的单个键执行add/removePatch不会触碰同 Aspect 中其他字段对整个customPropertiesMap 执行set替换同样只影响该 Mapname、description等字段保持原样。从 Python 侧看DatasetPatchBuilder中自定义属性操作的落点正是datasetPropertiesAspect 下的customProperties路径。见 dataset.pyclassmethod def _custom_properties_location(cls) - Tuple[str, PatchPath]: return DatasetProperties.ASPECT_NAME, (customProperties,)因此所有自定义属性 Patch 都会生成针对datasetProperties/customProperties路径的 MetadataChangeProposalMCPPatch 消息经 GMS 服务端应用后写入元数据存储。前置条件开始本教程前需要先完成以下准备工作部署 DataHub Quickstart按照 DataHub Quickstart 指南 启动本地 DataHub 实例并完成示例数据的摄取。确认目标数据集已存在在添加自定义属性之前必须确保目标数据集已经存在于你的 DataHub 实例中。如果尝试操作不存在的实体操作将失败。本指南使用示例数据摄取产生的fct_users_deleted数据集。完成示例数据摄取后fct_users_deleted数据集的自定义属性区域应该已经包含一个encoding属性值为utf-8。查看初始状态在操作之前可以使用 DataHub CLI 的datahub get命令读取datasetPropertiesAspect确认初始状态datahub get --urn urn:li:dataset:(urn:li:dataPlatform:hive,fct_users_deleted,PROD) --aspect datasetProperties { datasetProperties: { customProperties: { encoding: utf-8 }, description: table containing all the users deleted on a single day, tags: [] } }该 URN 的构成规则为urn:li:dataset:(urn:li:dataPlatform:platform,name,env)对应 Python 侧make_dataset_urn(platform, name, env)的生成逻辑见 mce_builder.py。操作一添加自定义属性Add添加操作向数据集追加一个或多个自定义属性不会影响已存在的属性。GraphQL 支持情况需要特别说明目前通过 GraphQL API 在 Dataset 上添加自定义属性暂不支持。更多各 API 能力差异请参考 DataHub API 能力对比表。因此以下实战将分别使用 Python SDK 与 Java SDK 完成。方式一Python SDK推荐完整示例代码位于 dataset_add_custom_properties_patch.pyfrom datahub.emitter.mce_builder import make_dataset_urn from datahub.ingestion.graph.client import DataHubGraph, DataHubGraphConfig from datahub.specific.dataset import DatasetPatchBuilder # Create DataHub Client datahub_client DataHubGraph(DataHubGraphConfig(serverhttp://localhost:8080)) # Create Dataset URN dataset_urn make_dataset_urn(platformhive, namefct_users_created, envPROD) # Create Dataset Patch to Add Custom Properties patch_builder DatasetPatchBuilder(dataset_urn) patch_builder.add_custom_property(cluster_name, datahubproject.acryl.io) patch_builder.add_custom_property(retention_time, 2 years) patch_mcps patch_builder.build() # Emit Dataset Patch for patch_mcp in patch_mcps: datahub_client.emit(patch_mcp)示例的关键步骤创建客户端DataHubGraph(DataHubGraphConfig(serverhttp://localhost:8080))指向本地 GMS 服务构造 URNmake_dataset_urn(platformhive, name..., envPROD)生成标准 Dataset URN构建 PatchDatasetPatchBuilder(dataset_urn)后连续调用add_custom_property(key, value)发送build()生成一组 MCP Patch 消息逐条通过datahub_client.emit()发送。方式二Java SDK完整示例代码位于 DatasetCustomPropertiesAdd.javaMetadataChangeProposal datasetPropertiesProposal new DatasetPropertiesPatchBuilder() .urn(UrnUtils.toDatasetUrn(hive, fct_users_deleted, PROD)) .addCustomProperty(cluster_name, datahubproject.acryl.io) .addCustomProperty(retention_time, 2 years) .build(); String token ; RestEmitter emitter RestEmitter.create(b - b.server(http://localhost:8080).token(token)); try { FutureMetadataWriteResponse response emitter.emit(datasetPropertiesProposal); System.out.println(response.get().getResponseContent()); } catch (Exception e) { log.error(Failed to emit metadata to DataHub, e); throw e; } finally { emitter.close(); }Java 侧对应使用DatasetPropertiesPatchBuilder通过UrnUtils.toDatasetUrn(hive, fct_users_deleted, PROD)构造 URN并通过RestEmitterHTTP 方式默认指向http://localhost:8080发送构建好的 MCP。底层实现原理Python 侧add_custom_property的底层实现位于 custom_properties.pydef add_custom_property(self, key: str, value: str) - Self: aspect_name, path self._custom_properties_location() self._add_patch( aspect_name, add, path(*path, key), valuevalue, ) return self它生成的操作语义是针对datasetPropertiesAspect 的customProperties/key路径执行addPatch。由于 Patch 操作是路径级的因此只影响目标键encoding等既有属性不会被触碰。此外HasCustomPropertiesPatch还提供了add_custom_properties(dict)批量添加方法内部循环调用add_custom_property。添加后的预期结果执行上述代码后fct_users_deleted将新增cluster_name与retention_time两个属性且原有encoding保持不变。使用 CLI 验证datahub get --urn urn:li:dataset:(urn:li:dataPlatform:hive,fct_users_deleted,PROD) --aspect datasetProperties { datasetProperties: { customProperties: { encoding: utf-8, cluster_name: datahubproject.acryl.io, retention_time: 2 years }, description: table containing all the users deleted on a single day, tags: [] } }操作二同时添加与移除自定义属性Add Remove实际运维中经常需要一次调用既新增又清理。下面的代码在一次 Patch 中同时完成添加cluster_name、移除retention_time且不影响其他属性。Python SDK完整示例代码位于 dataset_add_remove_custom_properties_patch.pyfrom datahub.emitter.mce_builder import make_dataset_urn from datahub.ingestion.graph.client import DataHubGraph, DataHubGraphConfig from datahub.specific.dataset import DatasetPatchBuilder # Create DataHub Client datahub_client DataHubGraph(DataHubGraphConfig(serverhttp://localhost:8080)) # Create Dataset URN dataset_urn make_dataset_urn(platformhive, namefct_users_created, envPROD) # Create Dataset Patch to Add Remove Custom Properties patch_builder DatasetPatchBuilder(dataset_urn) patch_builder.add_custom_property(cluster_name, datahubproject.acryl.io) patch_builder.remove_custom_property(retention_time) patch_mcps patch_builder.build() # Emit Dataset Patch for patch_mcp in patch_mcps: datahub_client.emit(patch_mcp)Java SDK完整示例代码位于 DatasetCustomPropertiesAddRemove.javaMetadataChangeProposal datasetPropertiesProposal new DatasetPropertiesPatchBuilder() .urn(UrnUtils.toDatasetUrn(hive, fct_users_deleted, PROD)) .addCustomProperty(cluster_name, datahubproject.acryl.io) .removeCustomProperty(retention_time) .build();底层实现原理remove_custom_property的底层实现同样是路径级的removePatch见 custom_properties.pydef remove_custom_property(self, key: str) - Self: aspect_name, path self._custom_properties_location() self._add_patch( aspect_name, remove, path(*path, key), value{}, ) return selfadd与remove两个 Patch 可以共存于同一次build()结果中由服务端顺序应用从而实现单次调用内的增量增删。操作后的预期结果执行后cluster_name被添加、retention_time被移除encoding仍保留datahub get --urn urn:li:dataset:(urn:li:dataPlatform:hive,fct_users_deleted,PROD) --aspect datasetProperties { datasetProperties: { customProperties: { encoding: utf-8, cluster_name: datahubproject.acryl.io }, description: table containing all the users deleted on a single day, tags: [] } }操作三整体替换自定义属性Replace替换操作将当前自定义属性 Map 整体替换为全新的 Map。例如下面的代码将属性 Map 替换为仅包含cluster_name与retention_time执行后原有的encoding将被移除。注意该操作仅替换customPropertiesMapdatasetPropertiesAspect 中同级的name、description等字段不受影响。Python SDK完整示例代码位于 dataset_replace_properties.py。该示例同时演示了 REST 与 Kafka 两种 Emitter 的使用方式from typing import Union from datahub.configuration.kafka import KafkaProducerConnectionConfig from datahub.emitter.kafka_emitter import DatahubKafkaEmitter, KafkaEmitterConfig from datahub.emitter.mce_builder import make_dataset_urn from datahub.emitter.rest_emitter import DataHubRestEmitter from datahub.specific.dataset import DatasetPatchBuilder # Get an emitter, either REST or Kafka, this example shows you both def get_emitter() - Union[DataHubRestEmitter, DatahubKafkaEmitter]: USE_REST_EMITTER True if USE_REST_EMITTER: gms_endpoint http://localhost:8080 return DataHubRestEmitter(gms_servergms_endpoint) else: kafka_server localhost:9092 schema_registry_url http://localhost:8081 return DatahubKafkaEmitter( configKafkaEmitterConfig( connectionKafkaProducerConnectionConfig( bootstrapkafka_server, schema_registry_urlschema_registry_url ) ) ) dataset_urn make_dataset_urn(platformhive, namefct_users_created, envPROD) property_map_to_set { cluster_name: datahubproject.acryl.io, retention_time: 2 years, } with get_emitter() as emitter: for patch_mcp in ( DatasetPatchBuilder(dataset_urn) .set_custom_properties(property_map_to_set) .build() ): emitter.emit(patch_mcp) print(fReplaced custom properties on dataset {dataset_urn} as {property_map_to_set})该示例有两点值得注意替换语义使用set_custom_properties(property_map_to_set)传入完整的字典整体覆盖既有属性 Map双通道发送通过get_emitter()可以在 REST Emitter直连 GMS HTTP 接口与 Kafka Emitter经 Kafka 消息队列异步写入之间切换两者都可配合with上下文管理自动释放资源。REST 模式默认端点http://localhost:8080Kafka 模式需要 Kafka 服务默认localhost:9092与 Schema Registry默认http://localhost:8081。Java SDK完整示例代码位于 DatasetCustomPropertiesReplace.javaMapString, String customPropsMap new HashMap(); customPropsMap.put(cluster_name, datahubproject.acryl.io); customPropsMap.put(retention_time, 2 years); MetadataChangeProposal datasetPropertiesProposal new DatasetPropertiesPatchBuilder() .urn(UrnUtils.toDatasetUrn(hive, fct_users_deleted, PROD)) .setCustomProperties(customPropsMap) .build();底层实现原理set_custom_properties与add_custom_property的区别在于 Patch 路径前者直接对customProperties根路径执行add以整个 Map 作为值见 custom_properties.pydef set_custom_properties(self, custom_properties: Dict[str, str]) - Self: Sets the custom properties of the entity. This method replaces all existing custom properties with the given dictionary. aspect_name, path self._custom_properties_location() self._add_patch( aspect_name, add, pathpath, valuecustom_properties, ) return self由于 Patch 的目标是(customProperties,)这一整个路径服务端会用新 Map 整体覆盖旧 Map从而产生替换效果而同一 Aspect 中的name、description等路径未被触碰因此保持不变。操作后的预期结果执行后cluster_name与retention_time存在encoding不再出现datahub get --urn urn:li:dataset:(urn:li:dataPlatform:hive,fct_users_deleted,PROD) --aspect datasetProperties { datasetProperties: { customProperties: { cluster_name: datahubproject.acryl.io, retention_time: 2 years }, description: table containing all the users deleted on a single day, tags: [] } }利用自定义属性增强搜索与发现如前文所述customProperties字段在 CustomProperties.pdl 中带有Searchable注解/*通配路径、fieldType: TEXT、queryByDefault: true因此所有自定义属性默认进入搜索索引可直接用于过滤与发现。仓库提供了开箱即用的搜索过滤示例 search_filter_by_custom_property.py使用 DataHub Python SDK 的 Filter DSL 按自定义属性检索所有资产from datahub.sdk import DataHubClient from datahub.sdk.search_filters import FilterDsl as F client DataHubClient(serveryour_server, tokenyour_token) # search for all assets with a custom property my_custom_property set to my_value results client.search.get_urns( filterF.has_custom_property(my_custom_property, my_value) )这意味着一套写入属性 → 建立索引 → 过滤检索的完整闭环写入侧使用DatasetPatchBuilder维护属性检索侧使用F.has_custom_property(key, value)精准过滤帮助用户在大量数据集中快速定位目标资产。三种操作语义速查操作影响范围典型场景关键 APIAdd添加仅目标键追加属性补充新属性而不动已有属性add_custom_property(key, value)/addCustomPropertyAdd Remove增删组合仅涉及的键单次调用完成新增与清理add_custom_propertyremove_custom_propertyReplace替换整个customPropertiesMap属性清单需要整体重置set_custom_properties(dict)/setCustomProperties三种操作都不会影响datasetPropertiesAspect 中customProperties之外的字段如name、description。同时请注意以上基于 Dataset 的自定义属性操作目前均不支持 GraphQL API请优先使用 Python SDK 或 Java SDK或参考 DataHub API 能力对比表 选择合适通道。总结通过本指南你已掌握在 DataHub 中维护数据集自定义属性的完整方法理解了customProperties在datasetPropertiesAspect 中的模型位置与可搜索性设计使用 PythonDatasetPatchBuilder与 JavaDatasetPropertiesPatchBuilder分别实现添加、增删组合与整体替换三种语义并通过datahub getCLI 验证每一次操作结果。这套基于 MCP Patch 的增量写入机制既保证了操作的原子性与精确性也让你能够基于自定义属性构建更强大的数据搜索与发现能力。【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考