如何使用专门用于 GCP 的 TFX SDK 实现 Kubeflow“运行参数”？

Posted 2023-03-29

【中文标题】如何使用专门用于 GCP 的 TFX SDK 实现 Kubeflow“运行参数”？

【英文标题】：How do I implement the Kubeflow "Run Paramters" with the TFX SDK specialized for GCP?

【发布时间】：2021-01-12 07:52:27

【问题描述】：

我目前使用 Kubeflow 作为我的编排器。编排器实际上是托管在 GCP 上的 AI 平台管道的一个实例。如何使用 Tensorflow 扩展 SDK 创建运行时参数？我怀疑这是我应该使用的类，但是文档不是很有意义，也没有提供任何示例。 https://www.tensorflow.org/tfx/api_docs/python/tfx/orchestration/data_types/RuntimeParameter

如下图所示。

【参考方案1】：

例如，您想将模块文件位置作为运行时参数添加到 TFX 管道中的转换组件。

首先设置你的 setup_pipeline.py 并定义模块文件参数：

# setup_pipeline.py

from typing import Text
from tfx.orchestration import data_types, pipeline
from tfx.orchestration.kubeflow import kubeflow_dag_runner
from tfx.components import Transform

_module_file_param = data_types.RuntimeParameter(
    name='module-file',
    default=
    '/tfx-src/tfx/examples/iris/iris_utils_native_keras.py',
    ptype=Text,
)

接下来，定义一个函数，指定管道中使用的组件并传递参数。

def create_pipeline(..., module_file):
    # setup components:
    ...

    transform = Transform(
         ...
         module_file=module_file
      )
     ...

    components = [..., transform, ...]

    return pipeline.Pipeline(
          ...,
          components=components
    )

最后，设置 Kubeflow DAG 运行器，以便将参数传递给 create_pipeline 函数。有关更完整的示例，请参阅 here。

if __name__ == "__main__":

    # instantiate a kfp_runner
    ...

    kfp_runner = kubeflow_dag_runner.KubeflowDagRunner(
        ...
    )

    kfp_runner.run(
        create_pipeline(..., module_file=_module_file_param
      ))

然后您可以运行python -m setup_pipeline，它将生成指定管道配置的 yaml 文件，然后您可以将其上传到 Kubeflow GCP 接口。