DataHub Actions的hello_world动作执行失败，报PipelineConfig验证错误

执行datahub actions -c hello_world.py时出现的5个验证错误，核心原因是你的配置文件缺少PipelineConfig要求的必填字段：filter、transform、action.config、datahub、options。

正确配置示例 #

以下是符合要求的hello_world.py完整配置模板：

from datahub_actions.action.action import Action
from datahub_actions.pipeline.pipeline import PipelineConfig, SourceConfig, ActionConfig

# 自定义Hello World Action逻辑
class HelloWorldAction(Action):
    def __init__(self, config):
        super().__init__()

    def act(self, event, context):
        print(f"Hello World! Received event: {event}")

# 完整Pipeline配置，必须包含所有必填字段
config = PipelineConfig(
    name="hello_world",
    source=SourceConfig(
        type="kafka",
        config={
            "bootstrap.servers": "localhost:9092",
            "topic": "datahub-mce-ingestion"
        }
    ),
    filter={
        # 示例：仅处理数据集创建事件，可根据需求调整
        "event_type": "EntityCreatedEvent",
        "entity_type": "dataset"
    },
    transform={
        # 无需转换时使用"noop"类型
        "type": "noop"
    },
    action=ActionConfig(
        type="hello_world",
        config={
            # 即使无配置参数，也需保留空字典
        }
    ),
    datahub={
        "gms_server": "http://localhost:8080"
    },
    options={
        "dry_run": False
    }
)

# 注册自定义Action
from datahub_actions.plugin_registry import PluginRegistry
PluginRegistry.register_action("hello_world", HelloWorldAction)

必填字段说明 #

• filter：定义事件过滤规则，只处理符合条件的事件
• transform：事件转换逻辑，无转换需求时用noop类型占位
• action.config：自定义Action的配置参数，即使无参数也需保留空字典
• datahub：配置DataHub GMS服务地址
• options：Pipeline全局选项，如是否启用干运行模式

调整配置后重新执行命令即可解决报错。

内容的提问来源于stack exchange，提问作者Vijay Jangir