DOC_INDEX
THEME
DOCS/Use Cases/Modular Data Pipeline

Modular Data Pipeline

A multi-region ETL composed from three reusable modules, with parallel extraction, quality gates, alerting and guaranteed cleanup

A regional sales pipeline assembled from three reusable modules: a source (fetch one region), a quality gate (assert a batch is sane) and a notifier. The orchestrating workflow fans the source module out across three regions in parallel, gates each batch, aggregates in JavaScript, and wraps the whole run in catch/finally so failures alert and raw files always get cleaned up.

Everything on this page was executed against OrchStep v0.7.1 - both the happy path and the failure path.

What it demonstrates

CapabilityWhere
Reusable module unitsthree local modules, each a plain directory with orchstep.yml
One module, many callsthe source module runs 3x with different with: values
Parallel module fan-outextract_all runs all regions concurrently
Cross-language data flowshell JSON -> result.data_object -> JS aggregation
Tunable policy--var min_orders=... flows into the gate via with:
Alert + guaranteed cleanupstep-level catch/finally around the pipeline task

The modules

modules/sales-source/orchstep.yml - any directory with an orchstep.yml is a module; no manifest required:

name: sales-source
desc: "Fetch raw sales records for a region (reusable source module)"

defaults:
  region: "apac"
  out_dir: "/tmp/orchstep-etl"

tasks:
  fetch:
    desc: "Fetch one region's sales as JSON lines"
    steps:
      - name: pull
        func: shell
        do: |
          mkdir -p {{ vars.out_dir }}
          OUT={{ vars.out_dir }}/{{ vars.region }}.json
          case "{{ vars.region }}" in
            apac) printf '{"region":"apac","orders":124,"revenue":18230.50}' > $OUT ;;
            emea) printf '{"region":"emea","orders":98,"revenue":15110.25}' > $OUT ;;
            amer) printf '{"region":"amer","orders":210,"revenue":33999.99}' > $OUT ;;
            *)    echo "unknown region" ; exit 1 ;;
          esac
          cat $OUT
        outputs:
          file: "{{ vars.out_dir }}/{{ vars.region }}.json"
          orders: "{{ result.data_object.orders }}"
          revenue: "{{ result.data_object.revenue }}"

modules/quality-gate/orchstep.yml:

name: quality-gate
desc: "Data quality assertions (reusable gate module)"

defaults:
  min_orders: "1"

tasks:
  check:
    desc: "Fail the pipeline when a batch looks broken"
    steps:
      - name: rules
        func: assert
        args:
          conditions:
            # values arrive as strings via `with:` - compare numerically in JS
            - condition: 'Number(vars.orders) >= Number(vars.min_orders)'
              desc: "orders above minimum"
            - condition: 'Number(vars.revenue) > 0'
              desc: "revenue must be positive"
        outputs:
          passed: "true"

modules/notify/orchstep.yml:

name: notify
desc: "Send a message to a channel (demo: prints; swap for slack CLI in production)"

defaults:
  channel: "data-eng"
  prefix: "INFO"

tasks:
  send:
    steps:
      - name: emit
        func: shell
        do: 'echo "[{{ vars.prefix }}] #{{ vars.channel }}: {{ vars.message }}"'
        outputs:
          line: "{{ result.output }}"

The orchestrating workflow

name: regional-sales-pipeline
desc: "Parallel multi-region ETL with module composition, quality gates and cleanup"

defaults:
  out_dir: "/tmp/orchstep-etl"
  report_dir: "./reports"
  min_orders: "50"

modules:
  - name: source
    source: "./modules/sales-source"
    config:
      out_dir: "/tmp/orchstep-etl"
  - name: quality
    source: "./modules/quality-gate"
  - name: notifier
    source: "./modules/notify"
    config:
      channel: "data-eng"

tasks:
  pipeline:
    desc: "Extract all regions in parallel, validate, aggregate, report"
    steps:
      - name: extract_all
        desc: "Each region uses the same source module with a different `with:`"
        parallel:
          - name: fetch_apac
            module: source
            task: fetch
            with: { region: "apac" }
          - name: fetch_emea
            module: source
            task: fetch
            with: { region: "emea" }
          - name: fetch_amer
            module: source
            task: fetch
            with: { region: "amer" }

      - name: gate_apac
        module: quality
        task: check
        with:
          orders: "{{ steps.fetch_apac.orders }}"
          revenue: "{{ steps.fetch_apac.revenue }}"
          min_orders: "{{ vars.min_orders }}"
      - name: gate_emea
        module: quality
        task: check
        with:
          orders: "{{ steps.fetch_emea.orders }}"
          revenue: "{{ steps.fetch_emea.revenue }}"
          min_orders: "{{ vars.min_orders }}"
      - name: gate_amer
        module: quality
        task: check
        with:
          orders: "{{ steps.fetch_amer.orders }}"
          revenue: "{{ steps.fetch_amer.revenue }}"
          min_orders: "{{ vars.min_orders }}"

      - name: aggregate
        desc: "Combine the three regions in JavaScript"
        func: transform
        do: |
          const regions = [
            { name: "apac", orders: Number(steps.fetch_apac.orders), revenue: Number(steps.fetch_apac.revenue) },
            { name: "emea", orders: Number(steps.fetch_emea.orders), revenue: Number(steps.fetch_emea.revenue) },
            { name: "amer", orders: Number(steps.fetch_amer.orders), revenue: Number(steps.fetch_amer.revenue) },
          ];
          const totals = regions.reduce((a, r) => ({orders: a.orders + r.orders, revenue: a.revenue + r.revenue}), {orders: 0, revenue: 0});
          const top = regions.slice().sort((a, b) => b.revenue - a.revenue)[0].name;
          return { total_orders: totals.orders, total_revenue: Math.round(totals.revenue * 100) / 100, top_region: top };

      - name: write_report
        func: shell
        do: |
          mkdir -p {{ vars.report_dir }}
          cat > {{ vars.report_dir }}/daily.txt <<R
          regional sales report
          total_orders  = {{ steps.aggregate.total_orders }}
          total_revenue = {{ steps.aggregate.total_revenue }}
          top_region    = {{ steps.aggregate.top_region }}
          R
          cat {{ vars.report_dir }}/daily.txt
        outputs:
          report: "{{ vars.report_dir }}/daily.txt"

      - name: announce
        module: notifier
        task: send
        with:
          message: "pipeline OK: {{ steps.aggregate.total_orders }} orders, top={{ steps.aggregate.top_region }}"

  main:
    desc: "Run the pipeline with alerting and guaranteed cleanup"
    steps:
      # The pipeline runs inside one step so catch/finally wrap the whole
      # run (works on all versions; v0.8.0+ also supports task-level).
      - name: run_pipeline
        task: pipeline
        catch:
          - name: alert_failure
            module: notifier
            task: send
            with:
              prefix: "ALERT"
              channel: "data-eng-oncall"
              message: "regional sales pipeline FAILED - check the run log"
        finally:
          - name: cleanup_raw
            func: shell
            do: |
              rm -rf {{ vars.out_dir }}
              echo "raw files removed"

Run it

# happy path
orchstep run main

# failure path: raise the quality bar so every region fails the gate
orchstep run main --var min_orders=500

Verified results:

# happy path (exit 0)
[INFO] #data-eng: pipeline OK: 432 orders, top=amer
raw files removed

# failure path (exit 0 - failure was caught, alerted, and cleaned up)
[ALERT] #data-eng-oncall: regional sales pipeline FAILED - check the run log
raw files removed

Design notes

Modules are interfaces. The orchestrator says what happens (fetch three regions, gate them, aggregate); each module owns how. Swapping the demo case statement in the source module for a real API call changes nothing in the orchestrator.

with: values are strings. Module parameters arrive as strings, so the quality gate compares numerically in JavaScript (Number(vars.orders) >= Number(vars.min_orders)) instead of Go-template gt, which fails on mixed types.

catch/finally placement. Since v0.8.0 catch:/finally: also attach directly to tasks; this workflow wraps the pipeline in a task: step instead, which works on every version and is still the way to protect a module task call. Either way the failure path exits 0 - the alert ran, cleanup ran, and the error was considered handled.

Take it to production

Point the source module at your real APIs (func: http instead of the demo case), publish the modules to a git repo and import them with source: "git:https://..." plus a version constraint, and route the notifier to Slack. The orchestrator does not change.

< Environment PromotionMonorepo CI Gate >