以编程方式读取 kubeflow 管道的输出

Question

我使用 python 命令在 Kubeflow 上运行管道，例如：

client.create_run_from_pipeline_func(pipeline_function, arguments=params_dict[name], run_name=name)

它在 Kubeflow 管道上创建一个作业，我希望能够使用 python API 访问有关管道不同步骤的信息。

job.get({step_name}).get_custom_properties({property_name})

我可以通过在 Kubeflow 中打开运行并查看我感兴趣的管道步骤的自定义属性来做到这一点，但我想自动化此过程。你知道是否可以使用 python API 来做到这一点吗？

Answer 1

我使用此类从 kubeflow 运行中提取参数：

import json
from typing import Dict
from typing import List

from kfp_server_api.models.api_run_detail import ApiRunDetail


class PipelineResult:

    DATASET_PATH_NAME = "data-load-features-DATA_SET_PATH"

    def __init__(self, run_description: ApiRunDetail):
        self._run_description = run_description

    @property
    def workflow_manifest(self) -> dict:
        return json.loads(self._run_description.pipeline_runtime.workflow_manifest)

    @property
    def status(self) -> str:
        return self.workflow_manifest["status"]["phase"]

    @property
    def params(self) -> List[Dict]:
        params_list = []
        for k, v in self.workflow_manifest["status"]["nodes"].items():

            for params in v.get("inputs", {}).get("parameters", []):
                params_list.append({"node_name": k, **params})
        return params_list

    def get_param(self, param_name: str):
        for el in self.params:
            if el["name"] == param_name:
                return el["value"]

    @property
    def training_set_path(self):
        return self.get_param(self.DATASET_PATH_NAME)

    @property
    def run_name(self):
        return self.workflow_manifest["metadata"]["annotations"]["pipelines.kubeflow.org/run_name"]

    def as_dict(self):
        return {
            "status": self.status,
            "training_set_path": self.training_set_path,
            "run_name": self.run_name,
        }


client = kfp.Client()
api_response = client.list_runs(namespace='...', sort_by='created_at desc', page_size=30)
runs_descriptions = [client.get_run(run.id) for run in api_response.runs]
runs = pd.DataFrame([PipelineResult(el).as_dict() for el in runs_descriptions])