How to Locally Debug Your Kubeflow Pipelines Before Committing

The Kubeflow Debugging Pain

Kubeflow Pipelines is powerful for orchestrating ML workflows on Kubernetes—but the development cycle is painfully slow when debugging requires compiling a pipeline, uploading it to the Kubeflow cluster, creating a run, waiting for pods to schedule, and then discovering a typo in line 47 of your preprocessing code. Each iteration takes 5-15 minutes. For a pipeline with 8 stages and a subtle data transformation bug, you might spend 2-3 hours in this compile-upload-wait-fix loop before identifying the issue.

Local debugging eliminates this cycle entirely. By running pipeline stages locally—with the same data, same dependencies, and same logic as the Kubeflow deployment—you can iterate in seconds instead of minutes. This article provides a complete guide to locally debugging Kubeflow Pipelines, covering lightweight components, container-based local execution, data mocking, and a development workflow that mirrors production.

Phase 1: Pipeline Structure for Local Debugging

Separate Logic from Orchestration

The most important architectural decision for debuggable pipelines is separating business logic from Kubeflow orchestration. Pipeline stages should call pure Python functions that can run anywhere—not Kubeflow-specific APIs.

# BAD: Logic embedded in Kubeflow component decorator
import kfp
from kfp import dsl

@dsl.component(base_image="python:3.11")
def preprocess_data(input_path: str, output_path: str):
    # This logic is trapped inside the Kubeflow component
    # Cannot be tested or debugged locally without Kubeflow
    import pandas as pd
    df = pd.read_parquet(input_path)
    df = df.dropna()
    df["feature"] = df["col_a"] * df["col_b"]
    df.to_parquet(output_path)

# GOOD: Logic in standalone functions, orchestrated by Kubeflow
# src/preprocessing.py — Pure Python, no Kubeflow dependencies
import pandas as pd
from pathlib import Path

def preprocess_data(input_path: str, output_path: str, 
                    drop_nulls: bool = True) -> pd.DataFrame:
    """Preprocess raw data. Pure function — no Kubeflow dependencies."""
    df = pd.read_parquet(input_path)
    
    if drop_nulls:
        null_count = df.isnull().sum().sum()
        print(f"Dropping {null_count} null values")
        df = df.dropna()
    
    df["feature"] = df["col_a"] * df["col_b"]
    
    # Write output
    Path(output_path).parent.mkdir(parents=True, exist_ok=True)
    df.to_parquet(output_path)
    
    return df

# pipeline/definition.py — Kubeflow orchestration layer
from kfp import dsl
from src.preprocessing import preprocess_data

@dsl.component(
    base_image="python:3.11",
    packages_to_install=["pandas", "pyarrow"],
)
def preprocess_component(input_path: str, output_path: str):
    """Thin wrapper that calls the pure Python function."""
    from src.preprocessing import preprocess_data
    preprocess_data(input_path, output_path)

@dsl.pipeline(name="ml-training-pipeline")
def training_pipeline(raw_data_path: str):
    preprocess = preprocess_component(
        input_path=raw_data_path,
        output_path="/tmp/preprocessed/data.parquet",
    )
    # ... rest of pipeline

Phase 2: Local Execution Framework

Local Pipeline Runner

# local_runner.py — Run Kubeflow pipeline stages locally
import subprocess
import json
import os
import tempfile
from pathlib import Path
from typing import Dict, List, Optional
from dataclasses import dataclass, field

@dataclass
class LocalPipelineContext:
    """Track state across locally-executed pipeline stages."""
    work_dir: str
    artifacts: Dict[str, str] = field(default_factory=dict)
    parameters: Dict[str, any] = field(default_factory=dict)
    stage_results: Dict[str, dict] = field(default_factory=dict)
    
    def get_artifact_path(self, name: str) -> str:
        """Get the local path for a named artifact."""
        if name not in self.artifacts:
            path = os.path.join(self.work_dir, name)
            self.artifacts[name] = path
        return self.artifacts[name]

class LocalPipelineRunner:
    """
    Execute Kubeflow pipeline stages locally for debugging.
    Supports both function-based and container-based execution.
    """
    
    def __init__(self, work_dir: str = None):
        self.work_dir = work_dir or tempfile.mkdtemp(prefix="kfp-local-")
        self.context = LocalPipelineContext(work_dir=self.work_dir)
        print(f"Local pipeline work directory: {self.work_dir}")
    
    def run_stage_function(self, stage_name: str, func, **kwargs):
        """
        Run a pipeline stage as a local Python function call.
        Fastest option for debugging — no container overhead.
        """
        print(f"\n{'='*60}")
        print(f"Running stage: {stage_name}")
        print(f"{'='*60}")
        
        # Resolve artifact paths
        resolved_kwargs = {}
        for key, value in kwargs.items():
            if isinstance(value, str) and value.startswith("artifact:"):
                artifact_name = value.replace("artifact:", "")
                resolved_kwargs[key] = self.context.get_artifact_path(artifact_name)
            else:
                resolved_kwargs[key] = value
        
        print(f"Parameters: {json.dumps(resolved_kwargs, indent=2, default=str)}")
        
        # Execute the function
        try:
            result = func(**resolved_kwargs)
            self.context.stage_results[stage_name] = {
                "status": "success",
                "result": str(result)[:500] if result else None,
            }
            print(f"✅ Stage '{stage_name}' completed successfully")
            return result
        except Exception as e:
            self.context.stage_results[stage_name] = {
                "status": "failed",
                "error": str(e),
            }
            print(f"❌ Stage '{stage_name}' failed: {e}")
            raise
    
    def run_stage_container(self, stage_name: str, image: str,
                            command: List[str], **kwargs):
        """
        Run a pipeline stage in a local Docker container.
        Use this to verify container-specific issues (dependencies, env vars).
        """
        print(f"\n{'='*60}")
        print(f"Running stage (container): {stage_name}")
        print(f"Image: {image}")
        print(f"{'='*60}")
        
        # Build docker run command
        docker_cmd = [
            "docker", "run", "--rm",
            "-v", f"{self.work_dir}:/workspace",
            "-w", "/workspace",
        ]
        
        # Add environment variables
        for key, value in kwargs.items():
            if key.startswith("env_"):
                env_name = key.replace("env_", "").upper()
                docker_cmd.extend(["-e", f"{env_name}={value}"])
        
        docker_cmd.append(image)
        docker_cmd.extend(command)
        
        print(f"Command: {' '.join(docker_cmd)}")
        
        result = subprocess.run(
            docker_cmd, capture_output=True, text=True
        )
        
        print(result.stdout)
        if result.stderr:
            print(f"STDERR: {result.stderr}")
        
        if result.returncode != 0:
            raise RuntimeError(f"Container exited with code {result.returncode}")
        
        self.context.stage_results[stage_name] = {"status": "success"}
        print(f"✅ Stage '{stage_name}' completed in container")
    
    def debug_stage(self, stage_name: str, func, **kwargs):
        """
        Run a stage with pdb debugger attached.
        Use this to step through code line-by-line.
        """
        print(f"\n🐛 Starting debugger for stage: {stage_name}")
        print(f"   Use 'n' to step, 'c' to continue, 'p var' to print variables")
        
        resolved_kwargs = {}
        for key, value in kwargs.items():
            if isinstance(value, str) and value.startswith("artifact:"):
                artifact_name = value.replace("artifact:", "")
                resolved_kwargs[key] = self.context.get_artifact_path(artifact_name)
            else:
                resolved_kwargs[key] = value
        
        import pdb
        pdb.set_trace()
        
        return func(**resolved_kwargs)


# Example: Run the full pipeline locally
if __name__ == "__main__":
    from src.preprocessing import preprocess_data
    from src.feature_engineering import engineer_features
    from src.training import train_model
    from src.evaluation import evaluate_model
    
    runner = LocalPipelineRunner()
    
    # Stage 1: Preprocessing
    runner.run_stage_function(
        "preprocess",
        preprocess_data,
        input_path="data/raw/train.parquet",
        output_path="artifact:preprocessed_data",
    )
    
    # Stage 2: Feature Engineering
    runner.run_stage_function(
        "feature_engineering",
        engineer_features,
        input_path="artifact:preprocessed_data",
        output_path="artifact:features",
    )
    
    # Stage 3: Training (with debugger)
    # runner.debug_stage(
    #     "training",
    #     train_model,
    #     features_path="artifact:features",
    #     model_path="artifact:model",
    # )
    
    # Or run normally:
    runner.run_stage_function(
        "training",
        train_model,
        features_path="artifact:features",
        model_path="artifact:model",
    )
    
    # Stage 4: Evaluation
    runner.run_stage_function(
        "evaluation",
        evaluate_model,
        model_path="artifact:model",
        features_path="artifact:features",
    )
    
    print(f"\nAll stages completed. Artifacts in: {runner.work_dir}")

Phase 3: Data Mocking for Local Development

Synthetic Data Generation

# local_dev/mock_data.py — Generate mock data that matches production schemas
import pandas as pd
import numpy as np
from pathlib import Path
from typing import Dict

class MockDataGenerator:
    """Generate synthetic data that matches production schemas for local testing."""
    
    def generate_raw_customer_data(self, n_rows: int = 10000) -> pd.DataFrame:
        """Generate mock customer data matching production schema."""
        np.random.seed(42)
        
        return pd.DataFrame({
            "customer_id": [f"CUST_{i:06d}" for i in range(n_rows)],
            "signup_date": pd.date_range("2022-01-01", periods=n_rows, freq="h"),
            "age": np.random.randint(18, 80, n_rows),
            "income": np.random.lognormal(10.5, 0.8, n_rows).astype(int),
            "plan_type": np.random.choice(["basic", "pro", "enterprise"], n_rows, p=[0.6, 0.3, 0.1]),
            "country": np.random.choice(["US", "UK", "DE", "FR", "JP"], n_rows),
            "churned": np.random.choice([0, 1], n_rows, p=[0.85, 0.15]),
        })
    
    def generate_transactions(self, n_rows: int = 50000) -> pd.DataFrame:
        """Generate mock transaction data."""
        np.random.seed(43)
        
        return pd.DataFrame({
            "transaction_id": [f"TXN_{i:08d}" for i in range(n_rows)],
            "customer_id": [f"CUST_{np.random.randint(0, 10000):06d}" for _ in range(n_rows)],
            "transaction_date": pd.date_range("2023-01-01", periods=n_rows, freq="10min"),
            "amount": np.random.lognormal(3, 1, n_rows).round(2),
            "category": np.random.choice(["subscription", "addon", "overage"], n_rows),
        })
    
    def generate_all(self, output_dir: str = "data/mock/"):
        """Generate all mock datasets and save to disk."""
        output_path = Path(output_dir)
        output_path.mkdir(parents=True, exist_ok=True)
        
        customers = self.generate_raw_customer_data()
        customers.to_parquet(output_path / "raw_customers.parquet")
        print(f"Generated {len(customers)} customer records")
        
        transactions = self.generate_transactions()
        transactions.to_parquet(output_path / "raw_transactions.parquet")
        print(f"Generated {len(transactions)} transaction records")
        
        # Generate a small subset for fast iteration
        customers.head(1000).to_parquet(output_path / "raw_customers_small.parquet")
        transactions.head(5000).to_parquet(output_path / "raw_transactions_small.parquet")
        print("Generated small subsets for fast debugging")


# Use in local runner
if __name__ == "__main__":
    generator = MockDataGenerator()
    generator.generate_all()

Phase 4: Testing Pipeline Stages

Unit Tests for Pipeline Logic

# tests/test_pipeline_stages.py — Unit tests for each pipeline stage
import pytest
import pandas as pd
import numpy as np
from src.preprocessing import preprocess_data
from src.feature_engineering import engineer_features
from src.training import train_model
from src.evaluation import evaluate_model
import tempfile
import os

class TestPreprocessing:
    """Test preprocessing stage in isolation."""
    
    def test_drops_null_values(self, tmp_path):
        # Create input with nulls
        input_df = pd.DataFrame({
            "col_a": [1, 2, None, 4],
            "col_b": [5, None, 7, 8],
        })
        input_path = str(tmp_path / "input.parquet")
        output_path = str(tmp_path / "output.parquet")
        input_df.to_parquet(input_path)
        
        # Run preprocessing
        result = preprocess_data(input_path, output_path, drop_nulls=True)
        
        # Verify nulls are dropped
        assert result.isnull().sum().sum() == 0
        assert len(result) == 2
    
    def test_computes_feature_column(self, tmp_path):
        input_df = pd.DataFrame({
            "col_a": [2, 3, 4],
            "col_b": [5, 6, 7],
        })
        input_path = str(tmp_path / "input.parquet")
        output_path = str(tmp_path / "output.parquet")
        input_df.to_parquet(input_path)
        
        result = preprocess_data(input_path, output_path)
        
        assert "feature" in result.columns
        np.testing.assert_array_equal(result["feature"].values, [10, 18, 28])
    
    def test_handles_empty_dataframe(self, tmp_path):
        input_df = pd.DataFrame({"col_a": [], "col_b": []})
        input_path = str(tmp_path / "input.parquet")
        output_path = str(tmp_path / "output.parquet")
        input_df.to_parquet(input_path)
        
        result = preprocess_data(input_path, output_path)
        assert len(result) == 0
    
    def test_output_file_created(self, tmp_path):
        input_df = pd.DataFrame({"col_a": [1, 2], "col_b": [3, 4]})
        input_path = str(tmp_path / "input.parquet")
        output_path = str(tmp_path / "output.parquet")
        input_df.to_parquet(input_path)
        
        preprocess_data(input_path, output_path)
        assert os.path.exists(output_path)


class TestFeatureEngineering:
    """Test feature engineering stage."""
    
    def test_generates_expected_features(self, tmp_path):
        input_df = pd.DataFrame({
            "customer_id": ["C1", "C2"],
            "age": [25, 35],
            "income": [50000, 80000],
        })
        input_path = str(tmp_path / "input.parquet")
        output_path = str(tmp_path / "output.parquet")
        input_df.to_parquet(input_path)
        
        result = engineer_features(input_path, output_path)
        
        expected_features = ["age_log", "income_bracket", "age_income_ratio"]
        for feat in expected_features:
            assert feat in result.columns
    
    def test_no_data_leakage(self, tmp_path):
        """Ensure feature engineering doesn't use target variable."""
        input_df = pd.DataFrame({
            "customer_id": ["C1"],
            "age": [25],
            "income": [50000],
            "target": [1],  # Target should NOT be used in features
        })
        input_path = str(tmp_path / "input.parquet")
        output_path = str(tmp_path / "output.parquet")
        input_df.to_parquet(input_path)
        
        result = engineer_features(input_path, output_path)
        
        # Target should not appear in feature columns
        assert "target" not in [c for c in result.columns if c.startswith("feat_")]


class TestTraining:
    """Test model training stage."""
    
    def test_model_trains_successfully(self, tmp_path):
        # Create small training dataset
        np.random.seed(42)
        n = 100
        features = pd.DataFrame({
            "feature_1": np.random.randn(n),
            "feature_2": np.random.randn(n),
            "target": np.random.choice([0, 1], n),
        })
        features_path = str(tmp_path / "features.parquet")
        model_path = str(tmp_path / "model.pkl")
        features.to_parquet(features_path)
        
        model = train_model(features_path, model_path)
        
        assert model is not None
        assert os.path.exists(model_path)
    
    def test_model_produces_valid_probabilities(self, tmp_path):
        np.random.seed(42)
        n = 100
        features = pd.DataFrame({
            "feature_1": np.random.randn(n),
            "target": np.random.choice([0, 1], n),
        })
        features_path = str(tmp_path / "features.parquet")
        model_path = str(tmp_path / "model.pkl")
        features.to_parquet(features_path)
        
        model = train_model(features_path, model_path)
        
        # Verify predictions are valid probabilities
        X_test = features.drop(columns=["target"])
        probas = model.predict_proba(X_test)[:, 1]
        assert np.all(probas >= 0) and np.all(probas <= 1)

Phase 5: Local-to-Kubeflow Parity Testing

Verify Local and Kubeflow Produce Identical Results

# tests/test_local_kubeflow_parity.py
import subprocess
import json
import pandas as pd
import numpy as np
from pathlib import Path

def test_local_matches_kubeflow():
    """
    Verify that local execution produces the same results as Kubeflow.
    Run this after any pipeline logic change.
    """
    
    # 1. Run locally
    from local_runner import LocalPipelineRunner
    from src.preprocessing import preprocess_data
    
    local_runner = LocalPipelineRunner(work_dir="/tmp/parity-test-local")
    local_runner.run_stage_function(
        "preprocess",
        preprocess_data,
        input_path="data/test/fixed_input.parquet",
        output_path="artifact:preprocessed",
    )
    
    local_output = pd.read_parquet(
        local_runner.context.get_artifact_path("preprocessed")
    )
    
    # 2. Run in Kubeflow (pre-compiled pipeline)
    result = subprocess.run(
        ["python", "scripts/run_kubeflow_stage.py",
         "--stage", "preprocess",
         "--input", "gs://test-data/fixed_input.parquet",
         "--output", "gs://test-data/parity_output/"],
        capture_output=True, text=True,
    )
    
    assert result.returncode == 0, f"Kubeflow run failed: {result.stderr}"
    
    # 3. Download Kubeflow output
    subprocess.run([
        "gsutil", "cp", "gs://test-data/parity_output/data.parquet",
        "/tmp/parity-test-kubeflow/data.parquet"
    ])
    
    kubeflow_output = pd.read_parquet("/tmp/parity-test-kubeflow/data.parquet")
    
    # 4. Compare outputs
    assert local_output.shape == kubeflow_output.shape, \
        f"Shape mismatch: local {local_output.shape} vs kubeflow {kubeflow_output.shape}"
    
    # Compare numeric columns with tolerance
    for col in local_output.select_dtypes(include=[np.number]).columns:
        np.testing.assert_allclose(
            local_output[col].values,
            kubeflow_output[col].values,
            rtol=1e-6,
            err_msg=f"Column {col} differs between local and Kubeflow"
        )
    
    print("✅ Local and Kubeflow outputs match")

Phase 6: Development Workflow

Recommended Workflow

1. Write logic as pure functions: All pipeline logic in src/ with no Kubeflow dependencies
2. Write unit tests: Test each stage function with small mock data
3. Run local pipeline: Execute the full pipeline locally with LocalPipelineRunner using mock data
4. Debug with pdb: Use runner.debug_stage() to step through failing stages
5. Test in container: Use runner.run_stage_container() to verify container-specific issues
6. Run parity test: Verify local output matches Kubeflow output
7. Compile and deploy: Only after local debugging passes, compile the Kubeflow pipeline and deploy

Conclusion

Local debugging transforms Kubeflow development from a slow, frustrating cycle into a fast, productive workflow. The key architectural decision is separating business logic from Kubeflow orchestration—pure Python functions that can run anywhere, wrapped in thin Kubeflow component decorators. The LocalPipelineRunner executes these functions locally with artifact tracking, mock data, and pdb debugging. Unit tests verify each stage in isolation. Parity tests ensure local and Kubeflow outputs match. Teams that adopt this workflow reduce pipeline debugging time by 80-90%—what took hours of compile-upload-wait cycles becomes minutes of local iteration.