如何为Datahub Ingestion Pipeline流转的数据集自动添加可编辑Structured Property

我在0.14.0.2版本的Datahub实例中，已通过CLI创建了结构化属性（Structured Property），但找不到将其批量应用到Ingestion Pipeline处理的所有数据集的方法。尝试使用add_dataset_properties转换器并指定结构化属性ID后，发现只是创建了不可修改的自定义属性，而非关联已定义的结构化属性。

请问有没有办法通过Pipeline自动为所有流转的数据集添加可在UI中修改值的结构化属性？还是只能在数据Ingestion完成后手动添加？

附已尝试的配置：

YAML配置 #

transformer:
- type: 'add_dataset_properties'
  config:
    semantics: PATCH
    add_properties_resolver_class: structured_property-resolver-file.PropertyResolverClass

Python解析器代码 #

from typing import Dict
from datahub.ingestion.transformer.add_dataset_properties import AddDatasetPropertiesResolverBase

class PropertyResolverClass(AddDatasetPropertiesResolverBase):
    def get_properties_to_add(self, entity_urn: str) -> Dict[str, str]:
        return {"structuredPropertyName": "desired default value"}

在Datahub 0.14.0.2版本中，add_dataset_properties转换器仅支持添加自定义属性，无法直接关联已创建的结构化属性。要实现自动为Pipeline中的数据集绑定可修改的结构化属性，需要通过调用Datahub的REST API完成，具体步骤如下：

1. 获取结构化属性的URN
   执行CLI命令列出已创建的结构化属性，拿到目标属性的propertyUrn：

datahub structured-property list

2. 自定义Transformer调用REST API
   编写自定义Transformer，在数据集Ingestion过程中调用Datahub的addEntityProperty接口，将结构化属性关联到数据集上。示例代码如下：

from typing import Optional
from datahub.ingestion.api.transformer import Transformer
from datahub.metadata.schema_classes import ChangeTypeClass
from datahub.rest import DatahubRestClient
from datahub.ingestion.graph.client import DatahubClientConfig

class AttachStructuredPropertyTransformer(Transformer):
    def __init__(self, config):
        self.client = DatahubRestClient(DatahubClientConfig(server=config["datahub_server"]))
        self.property_urn = config["structured_property_urn"]
        self.default_value = config.get("default_value", "")

    @classmethod
    def create(cls, config_dict: dict, ctx):
        return cls(config_dict)

    def transform(self, record):
        if record.entityType == "dataset" and record.entityUrn:
            # 调用API关联结构化属性
            self.client.add_entity_property(
                entity_urn=record.entityUrn,
                property_urn=self.property_urn,
                value=self.default_value,
                change_type=ChangeTypeClass.UPSERT
            )
        return record

3. 修改Ingestion配置
   在Pipeline配置中添加这个自定义Transformer，并配置Datahub服务地址和结构化属性URN：

transformer:
- type: 'custom'
  config:
    transformer_class: 'structured-property-attacher.AttachStructuredPropertyTransformer'
    datahub_server: 'http://your-datahub-server:8080'
    structured_property_urn: 'urn:li:structuredProperty:(urn:li:dataPlatform:hive,your-structured-property-name,string)'
    default_value: 'desired default value'

通过这种方式关联的结构化属性，会和手动添加的效果一致，支持在Datahub UI中修改属性值。

内容的提问来源于stack exchange，提问作者Frasher Gray