Kubeflow vs Argo vs Airflow: 4 Graph Execution Gotchas

Kubeflow’s DAG Isn’t Always a DAG

Most tutorials show you the happy path: define dependencies, watch the pipeline execute in order. But when you start adding conditional branches and loops to Kubeflow Pipelines, you’ll hit a wall the docs barely mention—cycles in your execution graph that don’t throw errors until runtime.

The core issue? Kubeflow’s conditional execution (dsl.Condition) and looping (dsl.ParallelFor) create dynamic subgraphs at runtime. If you’re not careful with how you wire dependencies, you can create cycles that look fine in the static compilation step but fail once Argo (the underlying executor) tries to resolve the actual execution order. I spent two days debugging a pipeline that compiled successfully but hung forever on a Pending status because a downstream task inadvertently depended on its own loop output through a condition check.

Here’s a simplified version of what broke:

import kfp
from kfp import dsl

@dsl.component
def preprocess_op(data: str) -> str:
    return f"processed_{data}"

@dsl.component
def check_quality_op(data: str) -> str:
    # Returns 'pass' or 'fail'
    return 'pass' if len(data) > 10 else 'fail'

@dsl.component
def retrain_op(data: str) -> str:
    return f"retrained_{data}"

@dsl.pipeline(name='conditional-loop-pipeline')
def my_pipeline(input_data: str):
    preprocess_task = preprocess_op(data=input_data)

    with dsl.ParallelFor(items=['batch1', 'batch2', 'batch3']) as batch:
        quality_check = check_quality_op(data=preprocess_task.output)

        # This creates a hidden cycle: condition depends on loop item,
        # but retrain_task tries to feed back into the next iteration
        with dsl.Condition(quality_check.output == 'fail'):
            retrain_task = retrain_op(data=batch)
            # If you try to use retrain_task.output as input to preprocess_task,
            # you've created a cycle that Argo can't resolve

The fix is to ensure that any conditional reruns or reprocessing steps are isolated within their iteration scope. Don’t try to feed outputs back into the parent loop’s dependency chain. Instead, collect failures and handle them in a separate downstream task.

Argo’s Exit Handler Fires Before You Think

Argo Workflows has this nice feature called exit handlers—tasks that run after the main workflow completes, regardless of success or failure. Perfect for cleanup, sending notifications, or logging final state.

Except it doesn’t wait for all your pods to terminate.

I had a pipeline that trained a model, pushed it to a registry, then used an exit handler to log the final training metrics to MLflow. The exit handler ran, logged the metrics… and then I’d check MLflow and see incomplete data. Sometimes the GPU training pod hadn’t fully flushed its logs yet. Sometimes the model upload was still in progress when the exit handler fired.

The issue is that Argo considers a workflow “done” once all steps are done, but the underlying Kubernetes pods might still be in Terminating state for a few seconds. If your exit handler needs to read logs or artifacts from those pods, you’re racing against kubelet’s cleanup.

Workaround: add an explicit wait step before your exit handler that polls for pod readiness or artifact availability. Ugly, but it works:

import time
import kubernetes

def wait_for_pod_termination(pod_name: str, namespace: str = 'default', timeout: int = 60):
    """Wait until pod fully terminates or timeout."""
    k8s_client = kubernetes.client.CoreV1Api()
    start = time.time()

    while time.time() - start < timeout:
        try:
            pod = k8s_client.read_namespaced_pod(name=pod_name, namespace=namespace)
            if pod.status.phase == 'Succeeded' or pod.status.phase == 'Failed':
                # Pod exists but finished—wait for actual deletion
                time.sleep(2)
            else:
                time.sleep(1)
        except kubernetes.client.rest.ApiException as e:
            if e.status == 404:
                # Pod deleted, we're good
                return
            raise

    # Timeout hit, log warning but don't fail
    print(f"Warning: pod {pod_name} still exists after {timeout}s")

Not elegant, but I haven’t found a better way to ensure artifacts are fully written before the exit handler runs. Mechanical Keyboard with Quiet Switches helps when you’re debugging this at 1am and your neighbors are asleep.

Airflow’s TaskGroup Doesn’t Parallelize Like You’d Expect

Airflow 2.0 introduced TaskGroup to organize tasks visually. You’d think grouping tasks would give you some control over parallelism within the group—maybe limit concurrency per group or force sequential execution.

Nope. TaskGroup is purely cosmetic for the UI. It doesn’t change execution behavior at all.

If you want to control parallelism, you still need to use max_active_runs, max_active_tasks_per_dag, or the older pool feature. I migrated a pipeline from Kubeflow to Airflow (as I mentioned in Airflow → Kubeflow Pipelines: 3 Breaking Changes I Hit), assuming TaskGroup would let me limit concurrent preprocessing tasks. It didn’t. All 50 tasks fired at once, crashed the cluster, and I had to retrofit pool assignments:

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.task_group import TaskGroup
from datetime import datetime

def preprocess_batch(batch_id: int):
    # Expensive preprocessing
    pass

with DAG('parallel-preprocess', start_date=datetime(2026, 1, 1), schedule=None) as dag:
    with TaskGroup('preprocessing') as preprocess_group:
        for i in range(50):
            # TaskGroup doesn't limit concurrency—you need pool assignment
            PythonOperator(
                task_id=f'preprocess_{i}',
                python_callable=preprocess_batch,
                op_kwargs={'batch_id': i},
                pool='preprocess_pool',  # This actually limits concurrency
            )

You define the pool in Airflow’s admin UI or via config:

# airflow.cfg
[core]
default_pool_task_slot_count = 10  # Global default

# Or via CLI:
# airflow pools set preprocess_pool 5 "Limit preprocessing concurrency"

The mental model is: TaskGroup = folder in the UI. Pool = actual concurrency control. Don’t conflate the two.

Kubeflow’s Artifact Passing Breaks with Large Outputs

Kubeflow Pipelines passes data between components via artifacts stored in object storage (MinIO by default, or S3/GCS). Small outputs (a few KB of JSON) work fine. But if your component returns a 500MB DataFrame serialized to Parquet, you’ll hit timeouts and OOM errors that aren’t surfaced clearly.

The artifact handoff goes like this:

1. Component A writes output to /tmp/outputs/output_name/data
2. Kubeflow’s executor sidecar uploads that file to MinIO
3. Component B’s executor sidecar downloads it to /tmp/inputs/input_name/data
4. Component B reads the file

If step 2 or 3 times out (default is 60s), the entire pipeline fails with a cryptic Failed to download artifact error. Worse, if the file is larger than the pod’s memory limit, the sidecar OOMs and Kubernetes just reschedules the pod in a loop.

I hit this trying to pass a 2GB model checkpoint between training and evaluation components. The training pod had 4GB RAM, but the sidecar tried to load the entire file into memory during upload, OOM’d, and the pod restarted. Took me an hour to realize the training step itself had succeeded—the failure was in the artifact upload.

Solution: either increase pod memory limits or switch to passing artifact references instead of the artifacts themselves. For large files, write directly to S3/GCS and pass the URI as a string parameter:

import kfp
from kfp import dsl
import boto3
import pickle

@dsl.component(base_image='python:3.11', packages_to_install=['boto3'])
def train_large_model_op(output_uri: dsl.Output[str]):
    # Train model, get large checkpoint
    model = {'weights': [1.0] * 10000000}  # Simulate 2GB model

    # Upload directly to S3, bypass Kubeflow artifact system
    s3 = boto3.client('s3')
    checkpoint_path = 's3://my-bucket/checkpoints/model.pkl'
    s3.put_object(
        Bucket='my-bucket',
        Key='checkpoints/model.pkl',
        Body=pickle.dumps(model)
    )

    # Return URI as a string, not the artifact itself
    output_uri.path = checkpoint_path
    with open(output_uri.path, 'w') as f:
        f.write(checkpoint_path)

@dsl.component(base_image='python:3.11', packages_to_install=['boto3'])
def evaluate_model_op(model_uri: str):
    s3 = boto3.client('s3')
    obj = s3.get_object(Bucket='my-bucket', Key='checkpoints/model.pkl')
    model = pickle.loads(obj['Body'].read())
    # Evaluate...

This adds S3 SDK overhead but sidesteps the entire Kubeflow artifact size limitation. For really large models (10GB+), consider using model registries like MLflow or just mounting a persistent volume claim across pods.

When to Use Which

Here’s my take after running all three in production:

Use Kubeflow Pipelines if you’re already on Kubernetes, need GPU orchestration, and want tight integration with ML frameworks (TFX, PyTorch Lightning). The artifact system is nice for small-to-medium outputs, and the UI gives you good experiment tracking. But be ready to wrestle with Argo’s quirks and write a lot of YAML.

Use Argo Workflows directly if you need more control over execution—retries, exit handlers, complex conditionals. Kubeflow abstracts away some of Argo’s power in exchange for Python-native syntax. If your pipeline has a lot of branching logic or needs sophisticated error handling, raw Argo might be cleaner. The learning curve is steeper (you’re writing YAML specs instead of Python), but you won’t hit the dynamic DAG issues Kubeflow sometimes creates.

Use Airflow if your pipeline is mostly ETL and light compute, you have Python dependencies that change often, and you value operational maturity over K8s-native features. The scheduler is rock-solid, the community is huge, and debugging is easier (just attach to the worker pod and import pdb). The tradeoff is worse support for GPU jobs and no native artifact versioning—you’ll build that yourself on top of S3/GCS.

One thing I’m still figuring out: hybrid setups. Can you run Airflow as the orchestrator and trigger Kubeflow pipelines as tasks? Technically yes (via KubernetesPodOperator or HTTP calls to Kubeflow’s API), but the operational overhead might not be worth it unless you have teams split between data engineering (Airflow) and ML engineering (Kubeflow). I haven’t tested that at scale yet.

FAQ

Q: Can I run Kubeflow Pipelines without Kubernetes?

No. Kubeflow Pipelines is fundamentally a K8s-native tool—it compiles your Python pipeline into Argo CRDs (Custom Resource Definitions) that only run on Kubernetes. If you want local development, you can use Kind or Minikube to spin up a local cluster, but there’s no “standalone” mode. For non-K8s environments, stick with Airflow or Prefect.

Q: Why does my Argo pipeline show Pending forever?

Usually one of three reasons: (1) insufficient cluster resources (your pods are waiting for CPU/GPU allocation), (2) image pull errors (check kubectl describe pod <pod-name> for ImagePullBackOff), or (3) you’ve hit a dependency cycle as described earlier. Run argo get <workflow-name> to see which step is stuck, then kubectl logs on that pod to see the actual error.

Q: How do I pass large datasets between Airflow tasks without storing in XCom?

XCom has a size limit (usually 48KB in the default SQLite backend, 64KB in Postgres). For large data, write to S3/GCS/MinIO in task A, return the URI via XCom, then read from that URI in task B. Or use Airflow 2.4+’s Dataset feature to declare data dependencies explicitly—tasks produce/consume datasets (URIs), and Airflow tracks lineage without moving the data through the metadata DB.

What I’d Change Next Time

If I were starting a new ML pipeline today, I’d prototype in Airflow for the first two sprints—just to get the data flow right without fighting Kubernetes. Once the DAG is stable, I’d migrate GPU-heavy steps (training, inference) to Kubeflow components and trigger them from Airflow via KubernetesPodOperator. Best of both worlds: Airflow’s scheduler + Python flexibility for orchestration, Kubeflow’s artifact tracking + GPU support for compute.

The part I haven’t solved cleanly is versioning pipeline definitions across tools. Airflow stores DAGs as Python files in a Git repo (easy), Kubeflow compiles to YAML (also version-controllable), but Argo CRDs generated by Kubeflow are ephemeral and don’t map 1:1 to the source Python. If you need to reproduce an experiment from six months ago, you have to re-run the Kubeflow compilation step with the old SDK version, which is fragile. I’m curious if tools like DVC or Pachyderm solve this, but I haven’t tested them yet.