#使用可复用的 Kubeflow Pipeline 组件

下一版 Alauda AI 将为 Kubeflow Pipelines 2.16.1 打包来自镜像 pipelines-components 项目的可复用组件。该 KFP 版本将由 Alauda kfp-operator v26.3.1 和 DSPO v2.15.2 提供。任一安装都会预加载两个 beta 版 AutoGluon pipeline：

托管 pipeline 是完成训练运行的最短路径。你也可以克隆该镜像，并在需要自定义 pipeline DAG 时导入单独的组件。

#前提条件

输入 bucket 可以不同于为 KFP artifact store 配置的对象存储。下面的 Secret 用于对数据加载器进行输入 bucket 身份验证；KFP 会单独为其 launcher 提供 artifact-store 凭据。

#创建输入存储 Secret

在 pipeline 运行命名空间中创建该 Secret：

NS=my-pipeline-namespace

kubectl -n "$NS" create secret generic pipeline-input-s3 \
  --from-literal=AWS_ACCESS_KEY_ID='<access-key>' \
  --from-literal=AWS_SECRET_ACCESS_KEY='<secret-key>' \
  --from-literal=AWS_S3_ENDPOINT='https://s3.example.com' \
  --from-literal=AWS_DEFAULT_REGION='us-east-1'

请保留 AWS_S3_ENDPOINT 中的 URI scheme。该组件会将此值作为 endpoint URL 传递给 S3 client。

在开始运行之前，将训练文件上传到 bucket。例如，下面的 tabular 输入包含两个特征和一个回归标签：

feature_a,feature_b,price
1.0,2.0,10.5
2.0,3.0,20.5
3.0,4.0,30.5

在去除重复行、无效标签和其他不可用值之后，tabular 数据必须至少包含 100 行有效数据。

#运行托管 tabular pipeline

1. 打开 Kubeflow Pipelines，然后选择 Pipelines。
2. 搜索 autogluon-tabular-training-pipeline 并选择其最新版本。
3. 选择 Create run，选择包含该 Secret 的命名空间，并设置以下输入：

4. 启动运行，并在运行图中跟踪数据加载器和模型训练任务。

该 pipeline 会对源数据进行采样和拆分，训练并排序 AutoGluon 模型，然后重新拟合表现最好的 top_n 个模型。训练拆分数据会通过运行的 workspace PVC 共享。测试拆分、leaderboard、模型 predictor、指标以及生成的推理 notebook 都是持久化到已配置 artifact store 中的 KFP artifact。

#运行托管 time-series pipeline

选择 autogluon-timeseries-training-pipeline，并提供上文所述的通用存储输入。其 time-series 专用输入如下：

加载器会对 ID/timestamp 对进行去重，并为每个序列执行时间拆分。它会将工作中的训练拆分写入 workspace PVC，并将测试拆分和已训练模型写入 artifact store。

#基于镜像组合自定义 pipeline

kfp-components 包未发布到 PyPI。请克隆 Alauda 镜像，并在 Python 3.11 或更高版本的虚拟环境中安装它。将 SDK 锁定到 Alauda AI 发布的版本。将 <pipelines-components-repo-url> 替换为 Alauda 为该环境提供的仓库地址：

git clone <pipelines-components-repo-url>
cd pipelines-components

python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install 'kfp==2.16.1' 'kfp-kubernetes==2.16.1'
python -m pip install --no-deps -e .

此示例导入 tabular 数据加载器，并将其用于一个更小的自定义 pipeline。请将镜像替换为镜像到你集群中的 AutoML runtime：

from kfp import compiler, dsl, kubernetes
from kfp_components.components.data_processing.automl.tabular_data_loader import (
    automl_data_loader,
)

AUTOML_IMAGE = "<registry>/mlops/kubeflow/odh-automl:v1.11.0.post.1"


@dsl.pipeline(
    name="custom-tabular-loader",
    pipeline_config=dsl.PipelineConfig(
        workspace=dsl.WorkspaceConfig(
            size="12Gi",
            kubernetes=dsl.KubernetesWorkspaceConfig(
                pvcSpecPatch={"accessModes": ["ReadWriteOnce"]}
            ),
        )
    ),
)
def custom_tabular_loader(
    secret_name: str,
    bucket: str,
    object_key: str,
    label: str,
    task_type: str = "regression",
):
    loader = automl_data_loader(
        bucket_name=bucket,
        file_key=object_key,
        workspace_path=dsl.WORKSPACE_PATH_PLACEHOLDER,
        label_column=label,
        task_type=task_type,
    )
    loader.set_container_image(AUTOML_IMAGE)
    kubernetes.use_secret_as_env(
        loader,
        secret_name=secret_name,
        secret_key_to_env={
            "AWS_ACCESS_KEY_ID": "AWS_ACCESS_KEY_ID",
            "AWS_SECRET_ACCESS_KEY": "AWS_SECRET_ACCESS_KEY",
            "AWS_S3_ENDPOINT": "AWS_S3_ENDPOINT",
            "AWS_DEFAULT_REGION": "AWS_DEFAULT_REGION",
        },
    )


compiler.Compiler().compile(
    pipeline_func=custom_tabular_loader,
    package_path="custom-tabular-loader.yaml",
)

从 Kubeflow Pipelines UI 上传 custom-tabular-loader.yaml，或者使用 KFP SDK 提交它。在使用其他资产之前，请先查看镜像中的组件 README.md、metadata.yaml 和测试；该仓库包含具有不同外部服务要求的 alpha 和 beta 资产。

#故障排查

• 找不到 Secret： Secret 是按命名空间作用域生效的。请在与运行相同的命名空间中创建输入 Secret。
• Workspace PVC 一直处于 Pending： 配置一个支持 ReadWriteOnce 的默认存储类，或者请平台管理员配置 KFP workspace 存储类。
• 运行一直处于 Pending： 将集群可用资源与 前提条件 中的 CPU 和内存请求进行比较。请从 preset=speed 开始。
• 无法读取输入对象： 检查 endpoint scheme、bucket 和对象 key、凭据、region、网络策略以及 S3 server certificate。
• Pod 完成后缺少 artifact： pipeline 输入 Secret 未配置 KFP 持久化。请与平台管理员一起检查 KFP 安装的 artifact-store endpoint 和凭据。