How to Build a Training Pipeline with Vertex AI Pipelines (Kubeflow SDK, End to End) | 2026

What Is a Vertex AI Training Pipeline?

Vertex AI Pipelines is Google Cloud's serverless orchestrator for ML workflows, built on Kubeflow Pipelines v2 (and TFX). Instead of a monolithic training script, you decompose the workflow into components — data validation, preprocessing, training, evaluation, conditional registration — each running as its own containerized job. Vertex AI executes the DAG, passes typed artifacts between steps through Cloud Storage, and records every run in ML Metadata for lineage and comparison.

Think of it like a professional bakery's production line instead of one chef doing everything from memory. Each station (mixing, baking, quality check, packaging) is a component with defined inputs and outputs, the conveyor belt is the pipeline DAG, the quality inspector who rejects a bad batch before packaging is your evaluation gate, and the order log tracking every ingredient batch is ML Metadata.

Why Not Just Run a Training Script on a Cron Job?

The classic setup — a monolithic `train.py` on a scheduled VM or Cloud Run job — breaks down as soon as the workflow grows: no isolation between steps (a data bug kills the whole run with no partial recovery), no lineage (which dataset version produced the model in production?), no caching (every run retrains from scratch even when only evaluation code changed), and no quality gate (a degenerate model silently overwrites the good one in the registry).

KFP on Vertex AI gives each step its own container, retry policy, and cache key; artifacts flow through typed channels with automatic lineage; and a `dsl.If` evaluation gate makes model registration a decision, not an accident. Because it is serverless, there is no Kubeflow cluster to babysit. For applying the same rigor to infrastructure around ML workloads, see [GCP Organization Policies to Enable by Default](/tutorial/gcp-organization-policies-defaults-terraform-module), and for cost attribution of training spend see [Per-Team Cost Allocation on GCP](/tutorial/gcp-per-team-cost-allocation-labels-folders-billing-queries).

Prerequisites

• GCP project with billing enabled
• gcloud CLI v450.0+ and Python 3.10+ installed and authenticated
• `roles/aiplatform.admin` and `roles/storage.admin` on the project (or equivalent custom roles)
• A Cloud Storage bucket for the pipeline root and staging (e.g. `gs://$PROJECT_ID-vertex-pipelines`)
• Python packages: `kfp>=2.5`, `google-cloud-aiplatform>=1.60`, `scikit-learn`, `pandas`

Step-by-Step Guide

Step 1: Enable APIs and Create the Pipeline Infrastructure

Enable the Vertex AI and supporting APIs, create the pipeline-root bucket, and provision a dedicated service account that pipeline runs will execute as. Pipeline steps run as a service account, not as you — least-privilege setup here decides whether a compromised component can read one bucket or your entire project, and Terraform makes the setup reproducible across dev/staging/prod.

gcloud services enable aiplatform.googleapis.com \
  storage.googleapis.com \
  cloudscheduler.googleapis.com \
  --project=$PROJECT_ID

gcloud storage buckets create gs://$PROJECT_ID-vertex-pipelines \
  --location=europe-west1 --uniform-bucket-level-access

gcloud iam service-accounts create vertex-pipeline-runner \
  --display-name="Vertex AI Pipeline Runner" --project=$PROJECT_ID

Step 2: Define the Data Preparation Component

Write the first `@dsl.component`: load the source data (BigQuery or GCS CSV), split into train/test, and emit typed `Dataset` artifacts plus row-count metrics. Declaring outputs as typed artifacts (not string paths) is what enables lineage, UI visualization, and cache hits — a component that returns `Output[Dataset]` slots cleanly into the DAG and the ML Metadata graph.

from kfp import dsl
from kfp.dsl import Dataset, Input, Metrics, Output

@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas==2.2.2", "scikit-learn==1.5.1"],
)
def prepare_data(
    source_uri: str,
    train_dataset: Output[Dataset],
    test_dataset: Output[Dataset],
    metrics: Output[Metrics],
):
    import pandas as pd
    from sklearn.model_selection import train_test_split

    df = pd.read_csv(source_uri)
    train_df, test_df = train_test_split(
        df, test_size=0.2, random_state=42, stratify=df["target"]
    )
    train_df.to_csv(train_dataset.path, index=False)
    test_df.to_csv(test_dataset.path, index=False)
    metrics.log_metric("train_rows", len(train_df))
    metrics.log_metric("test_rows", len(test_df))

Step 3: Define the Custom Training Component

Add the training component that consumes the train Dataset artifact, fits the model, and emits a typed `Model` artifact plus training metrics. This keeps training logic versioned in the pipeline rather than buried in a notebook. A typed Model output is the hand-off contract — the evaluation step and the conditional registration step both consume it, and Vertex AI renders it in the lineage graph so you can trace every registered model back to its exact training run.

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

@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas==2.2.2", "scikit-learn==1.5.1", "joblib==1.4.2"],
)
def train_model(
    train_dataset: Input[Dataset],
    n_estimators: int,
    model: Output[Model],
    metrics: Output[Metrics],
):
    import joblib
    import pandas as pd
    from sklearn.ensemble import GradientBoostingClassifier

    df = pd.read_csv(train_dataset.path)
    X, y = df.drop(columns=["target"]), df["target"]

    clf = GradientBoostingClassifier(n_estimators=n_estimators, random_state=42)
    clf.fit(X, y)

    model.metadata["framework"] = "scikit-learn"
    model.metadata["n_estimators"] = n_estimators
    with open(model.path, "wb") as f:
        joblib.dump(clf, f)
    metrics.log_metric("train_score", clf.score(X, y))

Step 4: Define Evaluation and Wire the Pipeline DAG

Write the evaluation component (accuracy/AUC against the test set, emitted as a `Metrics` artifact and a scalar output), then assemble everything in a `@dsl.pipeline` with a `dsl.If` gate: only register the model when accuracy clears the threshold. The evaluation gate is the difference between an MLOps pipeline and a cron job — a bad data batch or hyperparameter regression can never reach the Model Registry, because registration is a pipeline decision made on measured metrics.

from kfp import dsl
from kfp.dsl import Dataset, Input, Metrics, Model

@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas==2.2.2", "scikit-learn==1.5.1", "joblib==1.4.2"],
)
def evaluate_model(
    test_dataset: Input[Dataset],
    model: Input[Model],
    metrics: Output[Metrics],
) -> float:
    import joblib
    import pandas as pd
    from sklearn.metrics import accuracy_score, roc_auc_score

    df = pd.read_csv(test_dataset.path)
    X, y = df.drop(columns=["target"]), df["target"]
    clf = joblib.load(model.path)

    acc = accuracy_score(y, clf.predict(X))
    auc = roc_auc_score(y, clf.predict_proba(X)[:, 1])
    metrics.log_metric("accuracy", acc)
    metrics.log_metric("roc_auc", auc)
    return acc

@dsl.pipeline(
    name="churn-training-pipeline",
    description="End-to-end churn model training with evaluation gate",
)
def churn_pipeline(
    source_uri: str,
    n_estimators: int = 200,
    accuracy_threshold: float = 0.80,
):
    prep = prepare_data(source_uri=source_uri)
    train = train_model(
        train_dataset=prep.outputs["train_dataset"],
        n_estimators=n_estimators,
    )
    eval_task = evaluate_model(
        test_dataset=prep.outputs["test_dataset"],
        model=train.outputs["model"],
    )
    with dsl.If(eval_task.output >= accuracy_threshold, name="accuracy-gate"):
        register = register_model(model=train.outputs["model"])

Step 5: Compile the Pipeline to YAML

Use `kfp.compiler` to turn the decorated Python functions into a portable pipeline spec (IR YAML) that Vertex AI can execute — compilation is where syntax, typing, and DAG structure errors surface. Compiled YAML is the deployable artifact. Compiling in CI on every commit catches pipeline-breaking changes before a scheduled run burns training budget on a syntax error.

from kfp import compiler

compiler.Compiler().compile(
    pipeline_func=churn_pipeline,
    package_path="churn_pipeline.yaml",
)
print("Compiled to churn_pipeline.yaml")

Step 6: Submit the Pipeline Run to Vertex AI

Submit the compiled spec as a `PipelineJob` with the dedicated service account, pipeline root, and parameter values — then stream its state from the SDK. Running as the dedicated service account (step 1) rather than your user credentials keeps audit trails clean and permissions stable; the pipeline root namespace keeps run artifacts isolated and cacheable.

from google.cloud import aiplatform

aiplatform.init(
    project="PROJECT_ID",
    location="europe-west1",
    staging_bucket="gs://PROJECT_ID-vertex-pipelines",
)

job = aiplatform.PipelineJob(
    display_name="churn-training-run",
    template_path="churn_pipeline.yaml",
    pipeline_root="gs://PROJECT_ID-vertex-pipelines/runs",
    parameter_values={
        "source_uri": "gs://PROJECT_ID-vertex-pipelines/data/churn.csv",
        "n_estimators": 200,
        "accuracy_threshold": 0.80,
    },
    enable_caching=True,
)

job.run(
    service_account="vertex-pipeline-runner@PROJECT_ID.iam.gserviceaccount.com",
    sync=True,
)
print("Final state:", job.state)

Step 7: Schedule Recurring Training Runs

Wrap the same compiled pipeline in a Vertex AI pipeline schedule (Cloud Scheduler under the hood) so retraining happens on a cadence — weekly, or aligned with data arrival. Scheduled retraining with an evaluation gate is the core of continuous training: fresh models are produced automatically, but only models that beat the quality bar ever reach the registry — no human babysitting, no bad-model roulette.

from google.cloud import aiplatform

aiplatform.init(project="PROJECT_ID", location="europe-west1")

job = aiplatform.PipelineJob(
    display_name="churn-training-scheduled",
    template_path="churn_pipeline.yaml",
    pipeline_root="gs://PROJECT_ID-vertex-pipelines/runs",
    parameter_values={
        "source_uri": "gs://PROJECT_ID-vertex-pipelines/data/churn-latest.csv",
        "n_estimators": 200,
        "accuracy_threshold": 0.80,
    },
    enable_caching=True,
)

schedule = job.create_schedule(
    display_name="churn-weekly-retrain",
    cron="0 6 * * 1",  # Mondays 06:00 UTC
    service_account="vertex-pipeline-runner@PROJECT_ID.iam.gserviceaccount.com",
)
print("Schedule created:", schedule.resource_name)

Step 8: Verify Lineage and Set Up Run Alerts

Confirm the run produced a complete lineage graph (dataset → model → metrics → registered model) and add alerting on pipeline failures so a broken retrain is noticed before stakeholders are. Lineage is your audit trail — when a production model misbehaves, ML Metadata answers 'which data, which code, which parameters' in one query. Failure alerts close the loop: scheduled pipelines fail silently by default.

# Inspect run state and lineage
gcloud ai pipeline-jobs list \
  --region=europe-west1 --project=$PROJECT_ID \
  --filter="displayName:churn-training" --limit=5

# Failed-run alert (log-based metric + alert policy)
gcloud logging metrics create vertex_pipeline_failures \
  --description="Failed Vertex AI pipeline runs" \
  --log-filter='resource.type="aiplatform.googleapis.com/PipelineJob"
    AND jsonPayload.state="PIPELINE_STATE_FAILED"' \
  --project=$PROJECT_ID

Verification & Health Check

Best Practices

• Typed Artifacts, Not String Paths
• One Responsibility per Component
• Always Gate Registration on Metrics
• Pin Images and Package Versions

Common Mistakes

• {"errorCode":"PERMISSION_DENIED (aiplatform)","symptoms":"Pipeline run fails at submission or first component with 403 on `aiplatform.googleapis.com` or GCS access.","rootCause":"The pipeline-runner service account lacks `roles/aiplatform.user` or bucket access, or the submitter lacks `roles/iam.serviceAccountUser` (actAs) on it.","fixCommand":"gcloud projects add-iam-policy-binding $PROJECT_ID \\\n --member=\"serviceAccount:vertex-pipeline-runner@$PROJECT_ID.iam.gserviceaccount.com\" \\\n --role=\"roles/aiplatform.user\"\n","code":"resource \"google_service_account_iam_member\" \"submitter_actas\" {\n service_account_id = google_service_account.vertex_pipeline_runner.name\n role = \"roles/iam.serviceAccountUser\"\n member = \"user:[email protected]\"\n}\n","language":"hcl","filename":"fix-iam.tf","prevention":"Codify both the runner's roles and every submitter's actAs binding in Terraform from day one."}
• {"errorCode":"MODULE_NOT_FOUND inside component","symptoms":"Component pod crashes with `ModuleNotFoundError` for a package that is installed locally.","rootCause":"KFP serializes only the function body — imports outside the function or undeclared packages are not available in the component container.","fixCommand":"python compile_pipeline.py # recompile after moving imports inside\n","code":"@dsl.component(\n base_image=\"python:3.11-slim\",\n packages_to_install=[\"pandas==2.2.2\"],\n)\ndef prepare_data(...):\n import pandas as pd # imports live INSIDE the function\n","language":"python","filename":"fix-imports.py","prevention":"Lint components in CI by compiling the pipeline on every commit; compilation surfaces missing declarations immediately."}
• {"errorCode":"CACHE_NOT_HIT / SLOW_RERUNS","symptoms":"Every scheduled run re-executes all components; run time and cost stay at maximum.","rootCause":"`enable_caching` is False, or a component input changes every run (e.g. a timestamp parameter), invalidating cache keys.","fixCommand":"gcloud ai pipeline-jobs describe $JOB_ID --region=europe-west1 \\\n --format=\"value(jobDetail.pipelineRunContext)\"\n","code":"job = aiplatform.PipelineJob(\n display_name=\"churn-training-run\",\n template_path=\"churn_pipeline.yaml\",\n pipeline_root=\"gs://PROJECT_ID-vertex-pipelines/runs\",\n parameter_values={\"source_uri\": SOURCE_URI, \"n_estimators\": 200},\n enable_caching=True,\n)\n","language":"python","filename":"fix-caching.py","prevention":"Keep parameters stable across scheduled runs; put run-varying values (dates) into data content, not pipeline inputs."}
• {"errorCode":"PIPELINE_JOB_STUCK_PENDING","symptoms":"Run sits in PENDING for tens of minutes with no component starting.","rootCause":"Usually quota exhaustion (concurrent pipeline tasks or vCPU quota in the region) or a scheduling-created run referencing a deleted service account.","fixCommand":"gcloud ai pipeline-jobs describe $JOB_ID --region=europe-west1 \\\n --format=\"yaml(jobDetail.state, jobDetail.error)\"\n","code":"# Check regional quotas before scaling schedules\ngcloud compute regions describe europe-west1 \\\n --format=\"table(quotas.metric, quotas.limit, quotas.usage)\"\n","language":"bash","filename":"fix-quota.sh","prevention":"Monitor pending-duration per schedule and request quota increases before adding parallel retraining cadences."}

Cost Analysis

References

• Vertex AI Pipelines overview
• Build a pipeline with the Kubeflow Pipelines SDK v2
• KFP SDK v2 reference (kfp.dsl, kfp.compiler)
• Schedule a pipeline run on Vertex AI
• Compare pipelines and manage ML metadata