TFX 1.12.0中如何在管道执行时获取RuntimeParameter的字符串真实值？

问题原因 #

升级至tfx==1.12.0后，RuntimeParameter的__str__方法返回JSON序列化内容，导致路径拼接失效；同时tfx.proto.PushDestination.Filesystem仅接受字符串类型，无法直接传入RuntimeParameter（它是运行时解析的占位符，而非实际字符串值）。

可行方案：自定义Pusher Executor #

通过自定义Executor，在运行时获取RuntimeParameter的实际值，动态构建推送路径，具体步骤如下：

1. 定义自定义Pusher Executor
   继承官方Pusher的Executor，在Do方法中从运行时上下文提取参数，构建正确的PushDestination：

from tfx.components.pusher import executor as pusher_executor
from tfx.proto import pusher_pb2

class CustomPusherExecutor(pusher_executor.Executor):
    def Do(self, input_dict, output_dict, exec_properties):
        # 从运行时属性中获取data_root的实际值
        data_root = exec_properties['data_root']
        # 拼接模型服务路径
        model_root = f"{data_root}/model"
        
        # 构建PushDestination
        push_destination = pusher_pb2.PushDestination(
            filesystem=pusher_pb2.PushDestination.Filesystem(base_directory=model_root)
        )
        # 将构建好的目标路径传入执行属性，交给父类处理推送逻辑
        exec_properties['push_destination'] = push_destination
        
        # 调用父类的Do方法完成推送
        super().Do(input_dict, output_dict, exec_properties)

2. 使用自定义Executor构建Pusher组件
   将RuntimeParameter作为执行属性传入Pusher，指定自定义Executor：

import tfx
from tfx.components import Pusher

# 定义RuntimeParameter
data_root = tfx.orchestration.data_types.RuntimeParameter(name='data-root', ptype=str)

# 构建Pusher组件
pusher = Pusher(
    model=trainer.outputs['model'],
    model_blessing=evaluator.outputs['blessing'],
    # 指定自定义Executor
    executor_spec=tfx.components.executor_spec.ExecutorClassSpec(CustomPusherExecutor),
    # 传入RuntimeParameter作为执行属性
    exec_properties={'data_root': data_root}
)

说明 #

这种方式符合TFX RuntimeParameter的设计逻辑：参数在Kubeflow运行时由外部传入，自定义Executor在执行阶段（而非组件构造阶段）解析参数并构建推送路径，完美适配动态参数场景。

内容的提问来源于stack exchange，提问作者Youcef Kacer