Define Argo Workflow templates for data processing

Introduction

When you process terabytes of training corpora, validate embedding dimensions across millions of vectors, or transform multimodal datasets through sequential cleaning stages, simple cron jobs and shell scripts fall apart. Argo Workflows models each pipeline step as a Kubernetes-native container execution, combining container isolation with a purpose-built workflow engine. Getting these primitives wrong leaves you with brittle YAML that drifts across environments and silently swallows step failures, corrupting downstream training data.

By the end you'll be able to construct Argo Workflow and WorkflowTemplate manifests programmatically in Python, choose appropriate retry policies and resource limits for data steps, and reason about how the controller maps templates to pods at runtime.

Key Terminology

• Workflow — A Kubernetes Custom Resource declaring a complete pipeline execution; this is the unit you kubectl create to run a pipeline once.
• Container Template — A template specifying image, command, args, env, and resource limits, executed as one Pod. The atomic unit of work in every Argo data pipeline.
• WorkflowTemplate — A reusable, namespace-scoped resource storing template definitions independently of any single execution, so multiple pipelines can call the same processing step without duplication.
• Retry Strategy — Configures retry count, which failure types trigger retries (OnError, OnFailure, Always), and backoff — critical when one step in a 30-step pipeline trips a transient data-source error.
• activeDeadlineSeconds — Workflow-level timeout that terminates runaway pipelines so a stuck embedding job can't hold a GPU node forever.

Concepts

Workflow, template, and pod — the three-layer mapping

An Argo Workflow declares a directed graph of steps. Each step references a template, and the most fundamental type is the container template. When the workflow controller hits one, it creates a Pod, monitors it to completion, captures exit code and logs, then decides whether to advance, retry, or fail (see Code Walkthrough).

This model gives you four properties that matter for GenAI data processing:

• Isolation — Each step runs in its own pod with its own filesystem and resource quota, preventing a memory-hungry tokenization step from starving downstream validation.
• Reproducibility — Steps reference container images pinned to a digest, so historical pipeline runs are deterministic.
• Observability — Argo captures duration, resource consumption, and exit status per pod.
• Clean-slate retries — Each retry creates a fresh pod with a clean filesystem, so data steps that leave corrupted partial outputs on failure never inherit dirty state from a previous attempt.

Workflow vs WorkflowTemplate

A Workflow is the one-shot execution; a WorkflowTemplate is the reusable definition. Promote container templates to WorkflowTemplate once a pipeline is stable so other pipelines can reference it by name. The two manifest types share the same templates schema — only kind, metadata, and how they're triggered differ.

Retry policy selection for data steps

OnError retries only when the pod itself fails (eviction, OOM, image pull errors) and is the right default for deterministic transforms — re-running a script that exited non-zero on a logic bug will just fail again. Always retries on application failures too, appropriate when upstream APIs (object stores, embedding services) flake transiently. Combine with exponential backoff so a downstream service outage doesn't get hammered.

Code Walkthrough

Now that you understand the three-layer Workflow → template → pod mapping, the snippets below put those abstractions to work: the first builds a one-shot Workflow showing generateName, retry policy, and resource ceilings; the second promotes the same shape to a reusable WorkflowTemplate with parameterized inputs and optional GPU scheduling.

Code snippetpython
1from typing import Any, Optional
2
3def build_data_processing_workflow(
4    dataset_name: str,
5    image: str,
6    cpu_limit: str = "2",
7    memory_limit: str = "4Gi",
8    retry_count: int = 3,
9    active_deadline_seconds: Optional[int] = 3600,
10) -> dict[str, Any]:
11    """Build an Argo Workflow manifest for one-shot dataset processing."""
12    return {
13        "apiVersion": "argoproj.io/v1alpha1",
14        "kind": "Workflow",
15        "metadata": {
16            "generateName": f"process-{dataset_name}-",
17            "labels": {"pipeline": "data-processing", "dataset": dataset_name},
18        },
19        "spec": {
20            "entrypoint": "process-dataset",
21            "activeDeadlineSeconds": active_deadline_seconds,
22            "retryStrategy": {
23                "limit": str(retry_count),
24                "retryPolicy": "OnError",
25                "backoff": {"duration": "30s", "factor": 2, "maxDuration": "10m"},
26            },
27            "templates": [{
28                "name": "process-dataset",
29                "container": {
30                    "image": image,
31                    "command": ["python", "process.py"],
32                    "args": [
33                        "--dataset", dataset_name,
34                        "--validate", "true",
35                        "--output-format", "parquet",
36                    ],
37                    "resources": {
38                        "requests": {"cpu": "500m", "memory": "1Gi"},
39                        "limits": {"cpu": cpu_limit, "memory": memory_limit},
40                    },
41                },
42            }],
43        },
44    }

Four decisions worth calling out: generateName lets concurrent runs coexist without collisions; retryPolicy: "OnError" retries only infrastructure failures rather than application bugs; requests vs limits separates the scheduling guarantee from the hard ceiling that prevents one pod from evicting neighbors; activeDeadlineSeconds kills runaway embedding jobs before they hold a GPU node indefinitely.

Code snippetpython
1def build_workflow_template(
2    template_name: str,
3    image: str,
4    parameters: list[dict],
5    cpu_limit: str = "4",
6    memory_limit: str = "8Gi",
7    gpu_limit: int = 0,
8) -> dict[str, Any]:
9    """Build a reusable Argo WorkflowTemplate with optional GPU scheduling."""
10    resources: dict[str, dict[str, str]] = {
11        "requests": {"cpu": "1", "memory": "2Gi"},
12        "limits": {"cpu": cpu_limit, "memory": memory_limit},
13    }
14    if gpu_limit > 0:
15        resources["limits"]["nvidia.com/gpu"] = str(gpu_limit)
16
17    args: list[str] = []
18    for p in parameters:
19        args.append(f"--{p['name']}")
20        args.append("{{" + f"inputs.parameters.{p['name']}" + "}}")
21
22    return {
23        "apiVersion": "argoproj.io/v1alpha1",
24        "kind": "WorkflowTemplate",
25        "metadata": {"name": template_name},
26        "spec": {
27            "entrypoint": "run",
28            "arguments": {"parameters": parameters},
29            "templates": [{
30                "name": "run",
31                "inputs": {"parameters": parameters},
32                "container": {
33                    "image": image,
34                    "command": ["python", "process.py"],
35                    "args": args,
36                    "resources": resources,
37                },
38                "retryStrategy": {
39                    "limit": "3",
40                    "retryPolicy": "Always",
41                    "backoff": {"duration": "1m", "factor": 2, "maxDuration": "15m"},
42                },
43            }],
44        },
45    }

Three differences from the one-shot Workflow: name is fixed rather than generateName because WorkflowTemplate resources are long-lived and referenced by name from other pipelines; nvidia.com/gpu is added only when gpu_limit > 0 so CPU-only workloads never land on accelerator nodes; the {{inputs.parameters.NAME}} placeholders are Argo's template variable syntax — the double curly braces must reach Argo verbatim and are constructed via string concatenation rather than f-strings to prevent Python from interpreting them.

You'll know it works when kubectl create -f workflow.yaml returns workflow.argoproj.io/process-<dataset>-xxxxx created and argo get @latest shows the entrypoint template advancing through Pending → Running → Succeeded; the WorkflowTemplate is confirmed when argo submit --from workflowtemplate/<name> -p dataset=my-corpus produces a child Workflow that reuses the stored template definition without resubmitting the full manifest.

Do's and Don'ts

Do's

1. ✓Do pin container images by digest — referencing image: my-processor@sha256:... makes historical pipeline runs reproducible and removes the silent-drift class of failures.
2. ✓Do set both requests and limits — requests give the scheduler a placement guarantee; limits stop one pod from starving its neighbors on the node.
3. ✓Do promote stable Workflows to WorkflowTemplate — once a pipeline is steady, name it so other pipelines and CronWorkflows reference it instead of duplicating the manifest.

Don'ts

1. ✗Don't use retryPolicy: Always for deterministic transforms — retrying a logic bug just wastes compute and delays the inevitable failure signal.
2. ✗Don't f-string Argo template variables — {{inputs.parameters.NAME}} must reach Argo verbatim; collapsing the braces in Python breaks parameter substitution at runtime.
3. ✗Don't omit activeDeadlineSeconds — without it, a stuck embedding step can hold a GPU node indefinitely while the rest of the pipeline waits.