Kubeflow Pipelines 기초: 컴포넌트, 아티팩트, 로컬 실행에서 클러스터 제출까지

"$HOME/.local/bin/uv" venv --python 3.11 .venv
"$HOME/.local/bin/uv" pip install --python .venv/bin/python "kfp==2.17.3" "scikit-learn==1.7.2"

import argparse
import csv
import os

from kfp import compiler, dsl, local
from kfp.dsl import Dataset, Input, Metrics, Model, Output


@dsl.component(base_image="python:3.11")
def add(a: int, b: int) -> int:
    return a + b


@dsl.pipeline(name="intro-add")
def add_pipeline(a: int = 2, b: int = 3):
    add(a=a, b=b)


@dsl.component(base_image="python:3.11")
def write_sample(data: Output[Dataset]):
    with open(data.path, "w", encoding="utf-8") as stream:
        stream.write("value\n1\n2\n3\n")


@dsl.component(base_image="python:3.11")
def summarize(data: Input[Dataset], metrics: Output[Metrics]) -> int:
    with open(data.path, encoding="utf-8") as stream:
        values = [int(line) for line in stream.read().splitlines()[1:]]
    total = sum(values)
    metrics.log_metric("count", len(values))
    metrics.log_metric("sum", total)
    print(f"count={len(values)}, sum={total}")
    return total


@dsl.pipeline(name="artifact-demo")
def artifact_pipeline():
    source = write_sample()
    summarize(data=source.outputs["data"])


@dsl.component(base_image="python:3.11")
def label(text: str) -> str:
    print(f"selected: {text}")
    return text


@dsl.component(base_image="python:3.11")
def plus_one(value: int) -> int:
    return value + 1


@dsl.component(base_image="python:3.11")
def sum_values(values: list) -> float:
    return float(sum(values))


@dsl.pipeline(name="control-demo")
def control_pipeline(score: float = 0.8):
    with dsl.If(score >= 0.9):
        label(text="high")
    with dsl.Elif(score >= 0.7):
        label(text="medium")
    with dsl.Else():
        label(text="low")
    with dsl.ParallelFor(items=[1, 2, 3]) as item:
        task = plus_one(value=item)
    sum_values(values=dsl.Collected(task.output))


@dsl.component(base_image="python:3.11", packages_to_install=["scikit-learn==1.7.2"])
def create_iris_data(data: Output[Dataset]):
    import csv
    import numpy as np
    from sklearn.datasets import load_iris
    from sklearn.model_selection import train_test_split

    features, labels = load_iris(return_X_y=True)
    train_x, test_x, train_y, test_y = train_test_split(
        features, labels, test_size=0.25, random_state=42, stratify=labels
    )
    with open(data.path, "w", newline="", encoding="utf-8") as stream:
        writer = csv.writer(stream)
        writer.writerow(["split", "f0", "f1", "f2", "f3", "target"])
        for split, rows, targets in (
            ("train", train_x, train_y),
            ("test", test_x, test_y),
        ):
            for row, target in zip(rows, targets):
                writer.writerow([split, *np.asarray(row).tolist(), int(target)])
    print("Iris rows: 112 train, 38 test")


@dsl.component(base_image="python:3.11", packages_to_install=["scikit-learn==1.7.2"])
def train_iris(data: Input[Dataset], model: Output[Model]):
    import csv
    import joblib
    import numpy as np
    from sklearn.linear_model import LogisticRegression

    rows = []
    with open(data.path, newline="", encoding="utf-8") as stream:
        for row in csv.DictReader(stream):
            if row["split"] == "train":
                rows.append(row)
    features = np.asarray([[float(row[f"f{i}"]) for i in range(4)] for row in rows])
    labels = np.asarray([int(row["target"]) for row in rows])
    estimator = LogisticRegression(max_iter=300).fit(features, labels)
    joblib.dump(estimator, model.path)
    print(f"trained rows: {len(rows)}")


@dsl.component(base_image="python:3.11", packages_to_install=["scikit-learn==1.7.2"])
def evaluate_iris(
    data: Input[Dataset], model: Input[Model], metrics: Output[Metrics]
) -> float:
    import csv
    import joblib
    import numpy as np
    from sklearn.metrics import accuracy_score

    rows = []
    with open(data.path, newline="", encoding="utf-8") as stream:
        for row in csv.DictReader(stream):
            if row["split"] == "test":
                rows.append(row)
    features = np.asarray([[float(row[f"f{i}"]) for i in range(4)] for row in rows])
    labels = np.asarray([int(row["target"]) for row in rows])
    accuracy = float(accuracy_score(labels, joblib.load(model.path).predict(features)))
    metrics.log_metric("accuracy", accuracy)
    print(f"accuracy: {accuracy:.3f}")
    return accuracy


@dsl.component(base_image="python:3.11")
def quality_gate(accuracy: float, baseline: float = 0.9):
    if accuracy < baseline:
        raise RuntimeError(f"accuracy {accuracy:.3f} is below baseline {baseline:.3f}")
    print(f"quality gate passed: accuracy={accuracy:.3f}, baseline={baseline:.3f}")


@dsl.pipeline(name="iris-training")
def iris_pipeline(baseline: float = 0.9):
    data = create_iris_data()
    model = train_iris(data=data.outputs["data"])
    evaluation = evaluate_iris(data=data.outputs["data"], model=model.outputs["model"])
    quality_gate(accuracy=evaluation.outputs["Output"], baseline=baseline)


@dsl.component(base_image="python:3.11", packages_to_install=["pyyaml==6.0.3"])
def report_mode() -> str:
    import os
    mode = os.environ.get("PRIMER_MODE", "not-set")
    print(f"PRIMER_MODE={mode}")
    return mode


@dsl.pipeline(name="task-config-demo")
def task_config_pipeline():
    task = report_mode()
    task.set_caching_options(enable_caching=False)
    task.set_retry(num_retries=1)
    task.set_cpu_limit("1")
    task.set_memory_limit("512Mi")
    task.set_env_variable("PRIMER_MODE", "local")


@dsl.container_component
def container_hello():
    return dsl.ContainerSpec(
        image="python:3.11-slim",
        command=["python", "-c"],
        args=["print('container component')"],
    )


@dsl.pipeline(name="container-component-demo")
def container_pipeline():
    container_hello()


def compile_pipelines():
    targets = (
        (add_pipeline, "intro-add.yaml"),
        (artifact_pipeline, "artifact-demo.yaml"),
        (control_pipeline, "control-demo.yaml"),
        (iris_pipeline, "iris-training.yaml"),
        (task_config_pipeline, "task-config-demo.yaml"),
        (container_pipeline, "container-component-demo.yaml"),
    )
    for pipeline, filename in targets:
        compiler.Compiler().compile(pipeline, package_path=filename)
        print(f"compiled: {filename}")


def run_local(pipeline):
    local.init(
        runner=local.SubprocessRunner(use_venv=False),
        pipeline_root="./local_outputs",
    )
    pipeline()


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "action",
        choices=("compile", "intro", "artifacts", "control", "iris", "config"),
    )
    parser.add_argument("--score", type=float, default=0.8)
    parser.add_argument("--baseline", type=float, default=0.9)
    args = parser.parse_args()
    if args.action == "compile":
        compile_pipelines()
    elif args.action == "intro":
        run_local(add_pipeline)
    elif args.action == "artifacts":
        run_local(artifact_pipeline)
    elif args.action == "control":
        local.init(
            runner=local.SubprocessRunner(use_venv=False),
            pipeline_root="./local_outputs",
        )
        control_pipeline(score=args.score)
    elif args.action == "iris":
        local.init(
            runner=local.SubprocessRunner(use_venv=False),
            pipeline_root="./local_outputs",
        )
        iris_pipeline(baseline=args.baseline)
    else:
        run_local(task_config_pipeline)


if __name__ == "__main__":
    main()

.venv/bin/python kfp_primer.py compile

compiled: intro-add.yaml
compiled: artifact-demo.yaml
compiled: control-demo.yaml
compiled: iris-training.yaml
compiled: task-config-demo.yaml
compiled: container-component-demo.yaml

# PIPELINE DEFINITION
# Name: intro-add
# Inputs:
#    a: int [Default: 2.0]
#    b: int [Default: 3.0]
components:
  comp-add:
    executorLabel: exec-add
    inputDefinitions:
      parameters:
        a:
          parameterType: NUMBER_INTEGER
        b:
          parameterType: NUMBER_INTEGER

.venv/bin/python kfp_primer.py intro

Output: 5
Pipeline 'intro-add' finished with status SUCCESS

.venv/bin/python kfp_primer.py artifacts

count=3, sum=6
Output: 6
metadata={'count': 3.0, 'sum': 6.0}
Pipeline 'artifact-demo' finished with status SUCCESS

local_outputs/artifact-demo-<실행 시각>/write-sample/data
local_outputs/artifact-demo-<실행 시각>/summarize/metrics

.venv/bin/python kfp_primer.py control

selected: medium
Output: 2
Output: 3
Output: 4
Output: 9.0
Pipeline 'control-demo' finished with status SUCCESS

.venv/bin/python kfp_primer.py iris

Iris rows: 112 train, 38 test
trained rows: 112
accuracy: 0.947
quality gate passed: accuracy=0.947, baseline=0.900
Pipeline 'iris-training' finished with status SUCCESS