Decoupled Distributed Pipelines: Remote Worker Registration in wpipe

Building data pipelines that scale across multiple servers or ephemeral containers often leads to brittle, tightly-coupled architectures. When pipelines run isolated on edge nodes or batch workers, centrally orchestrating and monitoring their lifecycle requires a robust registration handshake with automatic local fallback.

With wpipe, distributed coordination is built into the core pipeline engine via api_config:

from wpipe import Pipeline

def process_data(data: dict) -> dict:
    return {"result": data["value"] * 2, "status": "processed"}

api_config = {
    "base_url": "http://orchestrator-cluster.local:8418",
    "token": "worker_auth_token_secret",
}

# Pipeline with remote registration capability
pipeline = Pipeline(
    worker_name="feature_engineering_node_01",
    api_config=api_config,
    verbose=True,
)

pipeline.set_steps([
    (process_data, "Process Raw Batch", "v1.0"),
])

# Register with central orchestrator with graceful local fallback
try:
    worker = pipeline.worker_register("feature_engineering_node_01", "v1.0")
    if worker:
        pipeline.set_worker_id(worker.get("id"))
    result = pipeline.run({"value": 42})
    print(f"Distributed execution result: {result}")
except Exception as e:
    print(f"Orchestrator unavailable ({e}), falling back to isolated execution:")
    result = pipeline.run({"value": 42})
    print(f"Local fallback result: {result}")

Why Senior Engineers Build Pipelines with wpipe:

• Centralized Telemetry & Local Autonomy: Pipelines report execution states, errors, and metadata when online, but execute locally without crashing if connectivity drops.
• SQLite WAL Checkpointing: Complete forensic resilience with persistent checkpoints that avoid rerunning expensive calculations.
• Zero GIL Contention: Effortless switching between process, thread, and native asyncio execution with identical API signatures.

Star and explore the full framework on GitHub!