nika: v1
workflow:
  id: etl-quarantine
  description: "CSV batch → schema gate → quarantine the bad · aggregate the good"

vars:
  batch_csv: "./data/incoming/orders.csv"

tasks:
  # A deterministic empty fallback · the recover target when parsing dies.
  empty_batch:
    invoke:
      tool: "nika:jq"
      args: { input: [], expression: "." }

  raw:
    invoke:
      tool: "nika:read"
      args: { path: "${{ vars.batch_csv }}" }

  rows:
    with:
      csv: ${{ tasks.raw.output }}
    invoke:
      tool: "nika:convert"
      args:
        input: "${{ with.csv }}"
        from: csv
        to: json
        has_header: true
    on_error:
      recover: ${{ tasks.empty_batch.output }}    # malformed CSV → empty batch · pipeline lives

  check:
    with:
      rows: ${{ tasks.rows.output }}
    invoke:
      tool: "nika:validate"
      args:
        data: "${{ with.rows }}"
        format: json
        schema:
          type: array
          items:
            type: object
            required: [order_id, amount, currency]
            properties:
              order_id: { type: string }
              amount: { type: string }
              currency: { type: string, enum: [EUR, USD, GBP] }

  good:
    with:
      rows: ${{ tasks.rows.output }}
      valid: ${{ tasks.check.output.valid }}
    when: ${{ with.valid == true }}
    invoke:
      tool: "nika:jq"
      args:
        input: "${{ with.rows }}"
        expression: 'group_by(.currency) | map({currency: .[0].currency, orders: length, total: (map(.amount | tonumber) | add)})'

  quarantine:
    with:
      valid: ${{ tasks.check.output.valid }}
      errors: ${{ tasks.check.output.errors }}
    when: ${{ with.valid == false }}
    invoke:
      tool: "nika:write"
      args:
        path: "./data/quarantine/orders-rejected.json"
        content: "${{ with.errors }}"
        create_dirs: true

  report:
    with:
      totals: ${{ tasks.good.output }}
    when: ${{ with.totals != null && size(with.totals) > 0 }}   # good is SKIPPED (null) on the quarantine path · guard before size()
    invoke:
      tool: "nika:write"
      args:
        path: "./data/reports/daily-totals.json"
        content: "${{ with.totals }}"
        create_dirs: true

outputs:
  # Deliberately untyped · on the quarantine path `good` is SKIPPED and
  # this is null — a declared `type: array` would refuse the branch the
  # file exists to demonstrate.
  totals: ${{ tasks.good.output }}

{"a":{"graph_format":2,"workflow":"etl-quarantine","nodes":[{"id":"empty_batch","verb":"invoke","tool":"nika:jq","permits":["tool: nika:jq"]},{"id":"raw","verb":"invoke","tool":"nika:read","permits":["tool: nika:read"]},{"id":"rows","verb":"invoke","tool":"nika:convert","permits":["tool: nika:convert"]},{"id":"check","verb":"invoke","tool":"nika:validate","permits":["tool: nika:validate"]},{"id":"good","verb":"invoke","tool":"nika:jq","when":"${{ with.valid == true }}","permits":["tool: nika:jq"]},{"id":"quarantine","verb":"invoke","tool":"nika:write","when":"${{ with.valid == false }}","permits":["fs.write: ./data/quarantine/orders-rejected.json","tool: nika:write"]},{"id":"report","verb":"invoke","tool":"nika:write","when":"${{ with.totals != null && size(with.totals) > 0 }}","permits":["fs.write: ./data/reports/daily-totals.json","tool: nika:write"]}],"edges":[{"from":"check","to":"good","kind":"value"},{"from":"check","to":"quarantine","kind":"value"},{"from":"check","to":"quarantine","kind":"value"},{"from":"empty_batch","to":"rows","kind":"recovery"},{"from":"good","to":"report","kind":"value"},{"from":"raw","to":"rows","kind":"value"},{"from":"rows","to":"check","kind":"value"},{"from":"rows","to":"good","kind":"value"}]},"engine":"0.105.0"}

showcase · tier 2 · ops & support

ETL quarantine

A schema gate splits good rows from bad. Rejects land in a quarantine file and the pipeline keeps going.

“Re-run the whole night” turns into “fix three rows tomorrow morning”. · conformance-gated in nika-spec ↗ · re-proven at every push

The whole file

7 tasks · 5 waves · the plan falls out of the bindings, nothing is scheduled by hand.

t2-etl-quarantine.nika.yamlsource

1nika: v12workflow:3  id: etl-quarantine4  description: "CSV batch → schema gate → quarantine the bad · aggregate the good"56vars:7  batch_csv: "./data/incoming/orders.csv"89tasks:10  # A deterministic empty fallback · the recover target when parsing dies.11  empty_batch:12    ◆invoke:13      tool: "nika:jq"14      args: { input: [], expression: "." }1516  raw:17    ◆invoke:18      tool: "nika:read"19      args: { path: "${{ vars.batch_csv }}" }2021  rows:22    with:23      csv: ${{ tasks.raw.output }}24    ◆invoke:25      tool: "nika:convert"26      args:27        input: "${{ with.csv }}"28        from: csv29        to: json30        has_header: true31    on_error:32      recover: ${{ tasks.empty_batch.output }}    # malformed CSV → empty batch · pipeline lives3334  check:35    with:36      rows: ${{ tasks.rows.output }}37    ◆invoke:38      tool: "nika:validate"39      args:40        data: "${{ with.rows }}"41        format: json42        schema:43          type: array44          items:45            type: object46            required: [order_id, amount, currency]47            properties:48              order_id: { type: string }49              amount: { type: string }50              currency: { type: string, enum: [EUR, USD, GBP] }5152  good:53    with:54      rows: ${{ tasks.rows.output }}55      valid: ${{ tasks.check.output.valid }}56    when: ${{ with.valid == true }}57    ◆invoke:58      tool: "nika:jq"59      args:60        input: "${{ with.rows }}"61        expression: 'group_by(.currency) | map({currency: .[0].currency, orders: length, total: (map(.amount | tonumber) | add)})'6263  quarantine:64    with:65      valid: ${{ tasks.check.output.valid }}66      errors: ${{ tasks.check.output.errors }}67    when: ${{ with.valid == false }}68    ◆invoke:69      tool: "nika:write"70      args:71        path: "./data/quarantine/orders-rejected.json"72        content: "${{ with.errors }}"73        create_dirs: true7475  report:76    with:77      totals: ${{ tasks.good.output }}78    when: ${{ with.totals != null && size(with.totals) > 0 }}   # good is SKIPPED (null) on the quarantine path · guard before size()79    ◆invoke:80      tool: "nika:write"81      args:82        path: "./data/reports/daily-totals.json"83        content: "${{ with.totals }}"84        create_dirs: true8586outputs:87  # Deliberately untyped · on the quarantine path `good` is SKIPPED and88  # this is null — a declared `type: array` would refuse the branch the89  # file exists to demonstrate.90  totals: ${{ tasks.good.output }}91

nika inspect · engine 0.105.0 · vendored graph, never re-derived

8 declared edges · 5 waves · the plan falls out of the bindings

try itopen in the playground guidethe walkthrough · docs verbinvoke toolnika:jq toolnika:read toolnika:convert toolnika:validate toolnika:write