企业官网建设流程全解析

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（替换），同样只影响该 Map，name、description等字段保持原样。

从 Python 侧看，DatasetPatchBuilder中自定义属性操作的落点正是datasetPropertiesAspect 下的customProperties路径。见 dataset.py：

@classmethod def _custom_properties_location(cls) -> Tuple[str, PatchPath]: return DatasetProperties.ASPECT_NAME, ("customProperties",)

因此，所有自定义属性 Patch 都会生成针对datasetProperties/customProperties路径的 MetadataChangeProposal（MCP）Patch 消息，经 GMS 服务端应用后写入元数据存储。

前置条件

开始本教程前，需要先完成以下准备工作：

1. 部署 DataHub Quickstart：按照 DataHub Quickstart 指南 启动本地 DataHub 实例，并完成示例数据的摄取。
2. 确认目标数据集已存在：在添加自定义属性之前，必须确保目标数据集已经存在于你的 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.py：

from 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(server="http://localhost:8080")) # Create Dataset URN dataset_urn = make_dataset_urn(platform="hive", name="fct_users_created", env="PROD") # 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)

示例的关键步骤：

1. 创建客户端：DataHubGraph(DataHubGraphConfig(server="http://localhost:8080"))，指向本地 GMS 服务；
2. 构造 URN：make_dataset_urn(platform="hive", name="...", env="PROD")生成标准 Dataset URN；
3. 构建 Patch：DatasetPatchBuilder(dataset_urn)后连续调用add_custom_property(key, value)；
4. 发送：build()生成一组 MCP Patch 消息，逐条通过datahub_client.emit()发送。

方式二：Java SDK

完整示例代码位于 DatasetCustomPropertiesAdd.java：

MetadataChangeProposal 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 { Future<MetadataWriteResponse> 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，并通过RestEmitter（HTTP 方式，默认指向http://localhost:8080）发送构建好的 MCP。

底层实现原理

Python 侧add_custom_property的底层实现位于 custom_properties.py：

def 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), value=value, ) 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.py：

from 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(server="http://localhost:8080")) # Create Dataset URN dataset_urn = make_dataset_urn(platform="hive", name="fct_users_created", env="PROD") # 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.java：

MetadataChangeProposal 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.py）：

def 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 self

add与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将被移除。注意：该操作仅替换customPropertiesMap，datasetPropertiesAspect 中同级的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_server=gms_endpoint) else: kafka_server = "localhost:9092" schema_registry_url = "http://localhost:8081" return DatahubKafkaEmitter( config=KafkaEmitterConfig( connection=KafkaProducerConnectionConfig( bootstrap=kafka_server, schema_registry_url=schema_registry_url ) ) ) dataset_urn = make_dataset_urn(platform="hive", name="fct_users_created", env="PROD") 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(f"Replaced custom properties on dataset {dataset_urn} as {property_map_to_set}")

该示例有两点值得注意：

1. 替换语义：使用set_custom_properties(property_map_to_set)传入完整的字典，整体覆盖既有属性 Map；
2. 双通道发送：通过get_emitter()可以在 REST Emitter（直连 GMS HTTP 接口）与 Kafka Emitter（经 Kafka 消息队列异步写入）之间切换，两者都可配合with上下文管理自动释放资源。REST 模式默认端点http://localhost:8080；Kafka 模式需要 Kafka 服务（默认localhost:9092）与 Schema Registry（默认http://localhost:8081）。

Java SDK

完整示例代码位于 DatasetCustomPropertiesReplace.java：

Map<String, 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.py）：

def 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", path=path, value=custom_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(server="<your_server>", token="<your_token>") # search for all assets with a custom property "my_custom_property" set to "my_value" results = client.search.get_urns( filter=F.has_custom_property("my_custom_property", "my_value") )

这意味着一套"写入属性 → 建立索引 → 过滤检索"的完整闭环：写入侧使用DatasetPatchBuilder维护属性，检索侧使用F.has_custom_property(key, value)精准过滤，帮助用户在大量数据集中快速定位目标资产。

三种操作语义速查

三种操作都不会影响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