Airflow → Kubeflow Pipelines: 3 Breaking Changes I Hit

The migration broke in ways the docs never mentioned

Kubeflow Pipelines promises “Airflow but cloud-native.” The reality? I spent three days fixing silent failures that wouldn’t happen in Airflow.

The switch isn’t just about rewriting DAGs as Python functions. Three specific breaking changes caught me off-guard, and none of them showed up in the official migration guides. Here’s what actually breaks when you move from Airflow to Kubeflow, with the exact errors and fixes.

Task dependencies behave completely differently

In Airflow, if Task B depends on Task A, and A fails, B simply doesn’t run. The DAG stops, you get a notification, life goes on.

Kubeflow Pipelines doesn’t work this way.

When a component fails in Kubeflow, downstream components can still execute depending on how you structured the pipeline. I had a data validation step fail (bad schema in incoming CSV), but the training step ran anyway using stale cached data from a previous run. The model trained successfully, got pushed to production, and started serving predictions on week-old features.

The root cause: Kubeflow’s caching system. By default, components cache outputs based on input parameters and code hash. If your component has the same inputs as a previous successful run, Kubeflow reuses the cached output even if an upstream dependency just failed.

# Airflow: explicit dependency, B won't run if A fails
task_a = PythonOperator(task_id='validate', python_callable=validate_data)
task_b = PythonOperator(task_id='train', python_callable=train_model)
task_a >> task_b  # clear dependency

# Kubeflow: looks similar, behaves differently
@dsl.pipeline(name='ml-pipeline')
def ml_pipeline():
    validate_op = validate_component()
    train_op = train_component(validate_op.output)  # should wait, right?
    # Wrong. If train_component has cached output, it SKIPS validate_op

The fix: disable caching for critical validation steps.

@dsl.pipeline(name='ml-pipeline')
def ml_pipeline():
    validate_op = validate_component()
    validate_op.execution_options.caching_strategy.max_cache_staleness = "P0D"  # zero-day cache
    train_op = train_component(validate_op.output)

But even this isn’t bulletproof. If you want true Airflow-style “stop everything on failure,” you need explicit exit handlers:

from kfp import dsl

@dsl.pipeline(name='ml-pipeline')
def ml_pipeline():
    validate_op = validate_component()

    with dsl.Condition(validate_op.output == "success"):  # only proceed if validation passed
        train_op = train_component(validate_op.output)
        deploy_op = deploy_component(train_op.output)

This adds verbosity. Every critical checkpoint now needs an explicit condition check. In Airflow, the execution model handled this by default.

Environment variables don’t propagate like you expect

Airflow lets you set environment variables in airflow.cfg or via Airflow Variables, and tasks inherit them automatically. Secrets get injected via connections or the Secrets Backend, and everything just works.

Kubeflow runs each component in its own Kubernetes pod with an isolated environment. If you need a variable (API key, database URL, S3 bucket name), you must explicitly pass it to every component.

I migrated a pipeline that hit an internal feature store API. The Airflow version used AIRFLOW_VAR_FEATURE_STORE_URL from the environment. The Kubeflow version? Silent 404s because the container didn’t have that variable.

# Airflow: variables available everywhere
from airflow.models import Variable
feature_store_url = Variable.get("FEATURE_STORE_URL")  # works in any task

# Kubeflow: need explicit injection per component
from kfp import dsl
from kubernetes.client.models import V1EnvVar

@dsl.component
def fetch_features(api_url: str) -> str:
    import requests
    response = requests.get(f"{api_url}/features")
    return response.json()

# You MUST pass api_url explicitly
@dsl.pipeline(name='feature-pipeline')
def feature_pipeline(feature_store_url: str):  # no auto-loading from env
    fetch_op = fetch_features(api_url=feature_store_url)

For secrets, it gets worse. Kubeflow expects you to create Kubernetes Secrets manually and mount them as environment variables:

from kubernetes.client.models import V1EnvVar, V1EnvVarSource, V1SecretKeySelector

@dsl.component
def train_model():
    # component code here
    pass

@dsl.pipeline(name='training-pipeline')
def training_pipeline():
    train_op = train_model()
    train_op.container.add_env_variable(
        V1EnvVar(
            name='AWS_ACCESS_KEY_ID',
            value_from=V1EnvVarSource(
                secret_key_ref=V1SecretKeySelector(
                    name='aws-credentials',  # must exist in k8s namespace
                    key='access_key'
                )
            )
        )
    )

If you forget to create the aws-credentials secret in the Kubeflow namespace, the pod crashes with CreateContainerConfigError. Airflow would’ve thrown a clear error during DAG parsing. Kubeflow fails at runtime, after the pipeline starts.

I ended up writing a helper function to inject common secrets:

def inject_standard_secrets(task_op):
    """Add AWS, DB, and API secrets to a Kubeflow component."""
    secrets = [
        ('AWS_ACCESS_KEY_ID', 'aws-credentials', 'access_key'),
        ('AWS_SECRET_ACCESS_KEY', 'aws-credentials', 'secret_key'),
        ('DB_PASSWORD', 'postgres-creds', 'password'),
        ('API_KEY', 'internal-api', 'key'),
    ]
    for env_name, secret_name, secret_key in secrets:
        task_op.container.add_env_variable(
            V1EnvVar(
                name=env_name,
                value_from=V1EnvVarSource(
                    secret_key_ref=V1SecretKeySelector(
                        name=secret_name,
                        key=secret_key
                    )
                )
            )
        )
    return task_op

# Usage
train_op = train_model()
inject_standard_secrets(train_op)

This works, but now you’re managing Kubernetes secrets separately from your pipeline code. In Airflow, everything lived in one place.

Dynamic task generation is completely different (and limited)

Airflow’s dynamic task mapping is straightforward. You return a list, Airflow creates N parallel tasks.

# Airflow: generate tasks based on runtime data
@task
def get_experiment_ids():
    return [101, 102, 103, 104, 105]  # fetched from DB at runtime

@task
def run_experiment(experiment_id: int):
    # train model for this experiment
    pass

ids = get_experiment_ids()
run_experiment.expand(experiment_id=ids)  # creates 5 parallel tasks

Kubeflow Pipelines supports parallelism via dsl.ParallelFor, but it has a critical limitation: the loop items must be known at compile time or passed as a pipeline parameter. You can’t dynamically generate tasks based on intermediate component outputs.

# Kubeflow: this does NOT work
@dsl.component
def get_experiment_ids() -> list:
    return [101, 102, 103, 104, 105]  # runtime list

@dsl.component
def run_experiment(experiment_id: int):
    pass

@dsl.pipeline(name='experiments')
def experiment_pipeline():
    ids_op = get_experiment_ids()
    # FAILS: ParallelFor can't consume component output directly
    with dsl.ParallelFor(ids_op.output):  # TypeError at compile time
        run_experiment()

The workaround: use a pipeline parameter and pass the list externally.

@dsl.pipeline(name='experiments')
def experiment_pipeline(experiment_ids: list):  # must be passed at submit time
    with dsl.ParallelFor(experiment_ids):
        run_experiment(dsl.LOOP_ITEM)

# Submit the pipeline
from kfp import Client
client = Client()
experiment_ids = [101, 102, 103, 104, 105]  # fetch this outside Kubeflow
client.create_run_from_pipeline_func(
    experiment_pipeline,
    arguments={'experiment_ids': experiment_ids}
)

This breaks the encapsulation. In Airflow, the DAG was self-contained: it queried the database, got the list, and spawned tasks. In Kubeflow, you need external orchestration to fetch the list before submitting the pipeline.

For true dynamic task generation, you need a custom launcher component that compiles and submits a new pipeline:

@dsl.component(base_image='python:3.9', packages_to_install=['kfp==2.0.0'])
def dynamic_launcher():
    from kfp import Client, dsl

    # Fetch experiment IDs at runtime
    experiment_ids = fetch_from_database()  # your DB query here

    # Define a sub-pipeline
    @dsl.pipeline(name='sub-experiments')
    def sub_pipeline(experiment_ids: list):
        with dsl.ParallelFor(experiment_ids):
            run_experiment(dsl.LOOP_ITEM)

    # Submit the sub-pipeline
    client = Client()
    client.create_run_from_pipeline_func(
        sub_pipeline,
        arguments={'experiment_ids': experiment_ids}
    )

This works, but now you’re managing nested pipeline submissions. The Kubeflow UI shows the launcher as one run, and the sub-pipeline as a separate run. Debugging becomes harder because logs are split across multiple pipeline executions.

I haven’t found a clean solution for this. My best guess is that Kubeflow’s design philosophy assumes you know your task graph at compile time, which isn’t realistic for many ML workflows (hyperparameter sweeps, A/B test variants, multi-model ensembles).

When Kubeflow makes sense anyway

Despite these pain points, Kubeflow wins on infrastructure.

Airflow runs on a single scheduler (or a small cluster if you set up Celery/Kubernetes executors). For ML pipelines that need 10+ GPU nodes, Airflow’s executor becomes the bottleneck. Kubeflow offloads scheduling to Kubernetes, which already handles distributed workloads well.

If you’re running pipelines on GKE, EKS, or AKS, Kubeflow’s native integration with pod autoscaling and node pools is a clear win. I’d pick Kubeflow over Airflow when:

• You need to scale beyond 50+ parallel tasks regularly
• Your tasks require heterogeneous compute (some CPU-only, some GPU, some TPU)
• You’re already managing Kubernetes infrastructure and don’t want to add Airflow as a separate service
• You need fine-grained resource limits per task ($T_i$ uses $\leq 4$ GB RAM, $T_j$ uses $\leq 2$ vCPUs)

But if your pipeline is mostly sequential ETL with occasional fan-out (e.g., 5-10 parallel model training runs), Airflow’s simpler execution model saves you from Kubeflow’s rough edges.

What I’d do differently next time

If I were migrating Airflow to Kubeflow again, I’d:

1. Audit dynamic task usage first. If your Airflow DAGs heavily use expand() or dynamic task generation based on runtime queries, budget extra time for Kubeflow workarounds. Consider keeping those DAGs in Airflow.
2. Set up secret management before writing components. Create all Kubernetes secrets upfront. Document which secrets each component needs. Airflow lets you lazily add secrets; Kubeflow punishes you at runtime.
3. Disable caching by default, enable selectively. Start with max_cache_staleness = "P0D" globally. Only enable caching for expensive, deterministic components (e.g., downloading a 50GB dataset). I’ve debugged too many “why did this succeed when it should’ve failed” issues caused by stale cache hits.
4. Write integration tests that simulate failures. In Airflow, you can trust that downstream tasks won’t run if upstream fails. In Kubeflow, test this explicitly. Inject a failure in your validation component and confirm that training doesn’t use cached outputs.

And honestly? I’d keep a small Airflow instance running for orchestration tasks that need dynamic behavior. Use Airflow to query the database, generate experiment configurations, and submit Kubeflow pipelines with those configs. Hybrid orchestration isn’t elegant, but it avoids rewriting complex dynamic DAGs into Kubeflow’s more rigid model.

FAQ

Q: Can I run Kubeflow Pipelines without Kubernetes?

No. Kubeflow Pipelines is built on Kubernetes and requires a K8s cluster. If you don’t have Kubernetes infrastructure, the setup overhead is steep — you’re better off sticking with Airflow or trying Prefect/Dagster.

Q: Does Kubeflow support Airflow-style sensors and hooks?

Not natively. Kubeflow components are stateless Python functions that run once and exit. For polling external systems (S3 file arrival, API status checks), you’d write a component with a retry loop, which is messier than Airflow’s S3KeySensor. Some teams use Airflow for orchestration and trigger Kubeflow pipelines via the SDK.

Q: Is Kubeflow Pipelines faster than Airflow for the same workload?

Not automatically. Task startup overhead is often higher in Kubeflow because each component runs in a new Kubernetes pod (image pull, container init, etc.). For long-running tasks (30+ min model training), this overhead is negligible. For short tasks (<1 min), Airflow’s persistent workers are faster. Kubeflow wins on scale, not speed.

The verdict: migrate only if Kubernetes is already your life

If you’re already running production workloads on Kubernetes and your team is comfortable debugging pod failures, CrashLoopBackOff errors, and YAML indentation, Kubeflow Pipelines is worth the migration pain. The resource isolation and autoscaling are genuinely better than Airflow’s executors.

But if you’re a small team (5 or fewer people) running ML pipelines that don’t need massive parallelism, Airflow’s simpler mental model will save you hours of debugging. The three breaking changes I hit — caching semantics, environment variable injection, and dynamic task limitations — aren’t dealbreakers, but they’re also not documented in the “Why Kubeflow?” blog posts.

I’m still curious whether Kubeflow v2’s SDK improvements (better type hints, compiled YAML introspection) make debugging easier. And I haven’t tested Vertex AI Pipelines (Google’s managed Kubeflow) to see if it smooths over some of these rough edges. If you’ve migrated recently and hit different issues, I’d love to hear what broke for you.

Debugging pipeline caching issues at 2am? Grab some Monster Energy Zero Ultra — you’ll need the caffeine and the satisfaction of crushing an empty can when you finally find the root cause.