ML Pipelines with Kubeflow Pipelines – Automate & Orchestrate ML Workflows

Key Learnings

• Why ML pipelines are essential in production ML workflows.
• What orchestration is and why DAGs matter.
• Kubeflow Pipelines for building and managing ML workflows.
• Step‑based (Airflow) vs component‑based (Kubeflow) workflows.
• Install Kubeflow and run a simple pipeline.

What are ML Pipelines?

ML pipelines are structured workflows that automate and streamline model development, deployment, and maintenance.

− 100% +

Why DAGs Matter

Kubeflow in a Nutshell

Kubeflow = Kubernetes + ML Workflows. An open‑source platform to build, deploy, and scale ML on K8s.

− 100% +

− 100% +

How Kubeflow Works

Central Dashboard, SDK/CLI, Pipelines UI, Argo Workflows, Katib, TF/PyTorch jobs, Artifact stores, Serving via KServe, monitoring via Prom/Grafana.

− 100% +

Step vs Component (Airflow vs KFP)

Simple scikit‑learn Pipeline (KFP)

Requirements: Python 3.7+, KFP SDK, Docker, running Kubeflow.

pip install kfp scikit-learn pandas joblib

from kfp import dsl
from kfp.components import create_component_from_func

# Step 1: Preprocess

def preprocess():
    from sklearn.datasets import load_iris
    import pandas as pd
    import joblib
    iris = load_iris(as_frame=True)
    df = iris.frame
    df.to_csv('/tmp/iris.csv', index=False)
    joblib.dump(iris.target_names, '/tmp/target_names.pkl')
    print("Data preprocessed")

preprocess_op = create_component_from_func(
    preprocess,
    base_image='python:3.8-slim',
    packages_to_install=['scikit-learn', 'pandas', 'joblib']
)

# Step 2: Train

def train():
    import pandas as pd
    from sklearn.tree import DecisionTreeClassifier
    import joblib
    df = pd.read_csv('/tmp/iris.csv')
    X = df.drop(columns='target')
    y = df['target']
    model = DecisionTreeClassifier()
    model.fit(X, y)
    joblib.dump(model, '/tmp/model.pkl')
    print("Model trained")

train_op = create_component_from_func(
    train,
    base_image='python:3.8-slim',
    packages_to_install=['scikit-learn', 'pandas', 'joblib']
)

# Step 3: Evaluate

def evaluate():
    import pandas as pd
    import joblib
    from sklearn.metrics import accuracy_score
    df = pd.read_csv('/tmp/iris.csv')
    X = df.drop(columns='target')
    y = df['target']
    model = joblib.load('/tmp/model.pkl')
    acc = accuracy_score(y, model.predict(X))
    print(f"Accuracy: {acc}")

evaluate_op = create_component_from_func(
    evaluate,
    base_image='python:3.8-slim',
    packages_to_install=['scikit-learn', 'pandas', 'joblib']
)

@dsl.pipeline(
    name='Scikit-learn Iris Pipeline',
    description='A pipeline that trains and evaluates a decision tree on Iris data'
)
def iris_pipeline():
    preprocess_task = preprocess_op()
    train_task = train_op().after(preprocess_task)
    evaluate_task = evaluate_op().after(train_task)

from kfp.compiler import Compiler
Compiler().compile(iris_pipeline, 'iris_pipeline.yaml')

− 100% +

− 100% +

Artifacts Passing (Improved)

from kfp import dsl
from kfp.components import create_component_from_func, InputPath, OutputPath

# Step 1: Preprocess → outputs CSV

def preprocess(output_data_path: OutputPath(str), output_target_names: OutputPath(str)):
    from sklearn.datasets import load_iris
    import pandas as pd
    import joblib
    iris = load_iris(as_frame=True)
    df = iris.frame
    df.to_csv(output_data_path, index=False)
    joblib.dump(iris.target_names, output_target_names)

preprocess_op = create_component_from_func(
    preprocess,
    base_image='python:3.8-slim',
    packages_to_install=['scikit-learn', 'pandas', 'joblib']
)

# Step 2: Train → outputs model

def train(input_data_path: InputPath(str), output_model_path: OutputPath(str)):
    import pandas as pd
    from sklearn.tree import DecisionTreeClassifier
    import joblib
    df = pd.read_csv(input_data_path)
    X = df.drop(columns='target')
    y = df['target']
    model = DecisionTreeClassifier()
    model.fit(X, y)
    joblib.dump(model, output_model_path)

train_op = create_component_from_func(
    train,
    base_image='python:3.8-slim',
    packages_to_install=['scikit-learn', 'pandas', 'joblib']
)

# Step 3: Evaluate

def evaluate(input_data_path: InputPath(str), input_model_path: InputPath(str)):
    import pandas as pd
    import joblib
    from sklearn.metrics import accuracy_score
    df = pd.read_csv(input_data_path)
    X = df.drop(columns='target')
    y = df['target']
    model = joblib.load(input_model_path)
    acc = accuracy_score(y, model.predict(X))
    print(f"Accuracy: {acc}")

evaluate_op = create_component_from_func(
    evaluate,
    base_image='python:3.8-slim',
    packages_to_install=['scikit-learn', 'pandas', 'joblib']
)

@dsl.pipeline(
    name='Scikit-learn Iris Pipeline with Artifacts',
    description='Pipeline demonstrating artifact passing'
)
def iris_pipeline():
    preprocess_task = preprocess_op()
    train_task = train_op(
        input_data_path=preprocess_task.outputs['output_data_path']
    )
    evaluate_task = evaluate_op(
        input_data_path=preprocess_task.outputs['output_data_path'],
        input_model_path=train_task.outputs['output_model_path']
    )

from kfp.compiler import Compiler
Compiler().compile(iris_pipeline, 'iris_pipeline_artifact.yaml')