Orchestrate pipelines as Argo Workflows with Kafka triggers

Introduction

When you ship an embedding pipeline as a Python script on someone's laptop, you've shipped a demo, not a production system. The moment ingestion volume grows, the script misses runs, retries nothing, leaks memory across hours-long jobs, and silently drops documents while no one is watching — until the search index goes stale and users complain. Argo Workflows turns that script into a Kubernetes-native DAG with explicit dependencies, retry budgets, and resource limits; Kafka triggers turn it from a cron job into an event-driven system that reacts the instant new documents land. By the end of this lesson you'll be able to express an embedding pipeline as an Argo Workflow DAG, attach retry and resource policies per step, and wire a Kafka topic to a sensor so each upstream ingestion event submits a fresh workflow run automatically.

Key Terminology

• Argo Workflow — a Kubernetes custom resource that runs a DAG of containerized steps with per-step retries, timeouts, and resource limits. It is the unit of execution that replaces the laptop script in this lesson.
• DAG template — the directed-acyclic-graph spec inside a Workflow, listing tasks and their dependencies. It encodes which embedding stages run in parallel (e.g. embed vs. cleanup) and which must wait (indexing waits for embedding).
• EventSource — an Argo Events resource that subscribes to an external system such as a Kafka topic and emits internal events. It is how the pipeline learns that upstream ingestion just finished.
• Sensor — an Argo Events resource that matches incoming events against dependencies and submits a Workflow per match. It is the glue that turns a Kafka message into a pipeline run with the right parameters.
• retryStrategy — per-template configuration (limit, retryPolicy, backoff) that retries a failed step with exponential backoff. It is what keeps the pipeline alive across transient embedding-API outages.

Concepts

From script to Kubernetes-native DAG

An embedding pipeline decomposes into discrete stages: change detection, embedding generation, vector-store indexing, cleanup of deleted chunks, and a cost report. Expressed as an Argo Workflow, each stage becomes a containerized step in a DAG with explicit dependencies. Change detection runs first; embedding waits on it; indexing waits on embedding; cleanup runs in parallel with indexing because it touches a disjoint chunk-id set; the cost report fans in at the end. The DAG shape is what gives you parallelism for free and removes the need for hand-rolled coordination code (see Code Walkthrough).

Retry budgets and resource limits per step

A single retryStrategy per template lets each stage carry the retry budget it actually needs: embedding gets three retries with 30s→5m exponential backoff because external provider outages are common; change detection gets two retries because it only queries Postgres. resources.requests/resources.limits bound CPU and memory so a runaway container cannot starve the rest of the cluster, and activeDeadlineSeconds caps each step's wall-clock so a stuck run fails loudly instead of hanging forever.

Event-driven execution via Kafka

Scheduled runs (cron) waste compute on idle hours and lag fresh documents by up to the cron interval. The Argo Events stack flips this around: an EventSource subscribes to a Kafka topic (e.g. document-ingested), a Sensor matches on those events and submits a Workflow per match, extracting parameters from the JSON message body (e.g. the collection name) via {{.Input.body.collection}}. The ingestion service just publishes; the sensor handles workflow creation. Decoupling is what makes the system safe to scale — neither side needs to know how the other is deployed.

Code Walkthrough

The two snippets below demonstrate the concepts from above: the DAG-plus-retry template defines the pipeline itself, and the EventSource-plus-Sensor pair wires Kafka events to workflow submission.

Code snippetyaml
1apiVersion: argoproj.io/v1alpha1
2kind: Workflow
3metadata:
4  generateName: embedding-pipeline-
5spec:
6  entrypoint: embedding-dag
7  arguments:
8    parameters:
9      - name: source-collection
10        value: "documents"
11  templates:
12    - name: embedding-dag
13      dag:
14        tasks:
15          - name: change-detection
16            template: detect-changes
17            arguments:
18              parameters:
19                - name: collection
20                  value: "{{workflow.parameters.source-collection}}"
21          - name: embed-chunks
22            template: run-embeddings
23            dependencies: [change-detection]
24          - name: index-vectors
25            template: store-vectors
26            dependencies: [embed-chunks]
27          - name: cleanup-deleted
28            template: cleanup
29            dependencies: [change-detection]
30          - name: cost-report
31            template: generate-report
32            dependencies: [index-vectors, cleanup-deleted]
33
34    - name: run-embeddings
35      retryStrategy:
36        limit: 3
37        retryPolicy: OnFailure
38        backoff:
39          duration: "30s"
40          factor: 2
41          maxDuration: "5m"
42      activeDeadlineSeconds: 3600
43      container:
44        image: embedding-pipeline:latest
45        command: [python, /app/embed.py]
46        resources:
47          requests: {memory: "2Gi", cpu: "500m"}
48          limits:   {memory: "4Gi", cpu: "2000m"}
49
50    - name: detect-changes
51      retryStrategy:
52        limit: 2
53        retryPolicy: OnFailure
54      activeDeadlineSeconds: 600
55      inputs:
56        parameters:
57          - name: collection
58      container:
59        image: embedding-pipeline:latest
60        command: [python, /app/detect.py]
61        args: ["--collection", "{{inputs.parameters.collection}}"]
62        resources:
63          requests: {memory: "512Mi", cpu: "250m"}
64          limits:   {memory: "1Gi",   cpu: "500m"}

The DAG block expresses the five stages and their dependencies; embedding and cleanup both depend on change detection but not on each other, so Argo runs them in parallel. The two template blocks attach retry and resource policies per step — the embedding step gets the bigger budget because it calls external APIs that occasionally 5xx, while change detection gets a smaller, faster budget because it just hits Postgres.

Code snippetyaml
1apiVersion: argoproj.io/v1alpha1
2kind: EventSource
3metadata:
4  name: document-events
5spec:
6  kafka:
7    document-ingested:
8      url: kafka-broker:9092
9      topic: document-ingested
10      consumerGroup:
11        groupName: embedding-trigger
12      jsonBody: true
13---
14apiVersion: argoproj.io/v1alpha1
15kind: Sensor
16metadata:
17  name: embedding-trigger
18spec:
19  dependencies:
20    - name: doc-event
21      eventSourceName: document-events
22      eventName: document-ingested
23  triggers:
24    - template:
25        name: run-embedding
26        argoWorkflow:
27          operation: submit
28          source:
29            resource:
30              apiVersion: argoproj.io/v1alpha1
31              kind: Workflow
32              metadata:
33                generateName: embedding-pipeline-
34              spec:
35                entrypoint: embedding-dag
36                arguments:
37                  parameters:
38                    - name: source-collection
39                      value: "{{.Input.body.collection}}"

The EventSource subscribes to the document-ingested Kafka topic with jsonBody: true so individual JSON fields are addressable downstream. The Sensor declares a dependency on that event and submits a Workflow per match, pulling the collection field from the message body into the workflow's source-collection parameter via {{.Input.body.collection}} — so each ingestion event produces a workflow run scoped to exactly the collection that changed.

You'll know it works when publishing a JSON message like {"collection":"docs-prod"} to the document-ingested topic causes kubectl get workflows to show a new embedding-pipeline-<hash> run within a few seconds, and kubectl logs on its pods shows the --collection docs-prod argument propagated through.

Do's and Don'ts

Now that you have worked through the implementation, the practices below separate a durable approach from a fragile one.

Do's

1. ✓Do set per-step retryStrategy and activeDeadlineSeconds — transient embedding-API failures need bounded retries, and a deadline turns "stuck forever" into a loud failure you can alert on.
2. ✓Do extract Kafka payload fields into workflow parameters — using {{.Input.body.<field>}} keeps the ingestion service decoupled from the pipeline's internals.
3. ✓Do let Argo parallelise independent stages — declare dependencies only where they truly exist (cleanup does not need to wait for indexing), and the DAG runtime handles the rest.

Don'ts

1. ✗Don't reach for cron when an event will do — scheduled runs lag fresh documents and burn compute on idle intervals; Kafka triggers fire only when there is real work.
2. ✗Don't omit resources.limits — a runaway embedding step without a memory cap will get OOM-killed by the node, taking unrelated pods with it.
3. ✗Don't bake secrets or broker URLs into the Workflow YAML — read them from a ConfigMap or Secret so the same DAG runs across dev, staging, and prod without edits.