Databricks DE · Topic 7.2

Job Control Flow

Make a job dynamic: branch with if/else (condition_task), loop with for-each, pass data with task values, and run tasks conditionally with run_if rules. This deep dive pairs each with the bundle YAML.

condition_taskfor_each_tasktask valuesrun_if

The one-liner

Control flow turns a fixed task chain into a dynamic workflow: If/else (condition_task) branches on a condition, For-each (for_each_task) loops a task over a list, task values pass small data between tasks, and run_if controls when a task runs based on upstream outcomes.

🍳 Analogy: on top of the recipe (the job): "if the dough's too dry, add water" (if/else), "repeat for each tray" (for-each), a sticky note passed between cooks (task values), and "only plate if the main didn't burn" (run-if).

If/else — drag the row count

▸ condition: row_count >= 100 → publish, else → quarantine

row_count (task value): 180

✅ Publish to gold

outcome: true

🚧 Quarantine + alert

outcome: false

1 · Task values — pass small data between tasks

# upstream task ("ingest"): publish a value
dbutils.jobs.taskValues.set(key="row_count", value=1234)
# downstream task: read it
n = dbutils.jobs.taskValues.get(taskKey="ingest", key="row_count", default=0)

Reference a value in a condition/param with {{tasks.ingest.values.row_count}}. Values are small (counts/flags) — for big data, write a table and pass a pointer.

2 · If/else — condition_task (bundle YAML)

tasks:
  - task_key: gate
    condition_task:
      op: GREATER_THAN              # EQUAL_TO, LESS_THAN, GREATER_THAN_OR_EQUAL, ...
      left: "{{tasks.ingest.values.row_count}}"
      right: "100"
  - task_key: publish_gold           # runs only when gate == true
    depends_on: [{ task_key: gate, outcome: "true" }]
    notebook_task: { notebook_path: ../src/publish.py }
  - task_key: quarantine             # runs only when gate == false
    depends_on: [{ task_key: gate, outcome: "false" }]
    notebook_task: { notebook_path: ../src/quarantine.py }

3 · For-each — for_each_task (bundle YAML)

tasks:
  - task_key: ingest_all
    for_each_task:
      inputs: '["sales","orders","users"]'   # or a task value / job param
      concurrency: 4                       # bound the parallelism
      task:
        task_key: ingest_one
        notebook_task:
          notebook_path: ../src/ingest_table.py
          base_parameters: { table: "{{input}}" }   # current item

4 · The four tools — click each

condition_task — branch on a condition comparing task values / job params (op + left + right); downstream tasks attach to the true/false outcome.

for_each_task — run a nested task once per item in inputs, passing {{input}} each iteration; concurrency bounds parallelism.

Task values — dbutils.jobs.taskValues.set/get pass small data between tasks without external storage.

run_if — choose when a task runs vs its dependencies (see table below). Use ALL_DONE / AT_LEAST_ONE_FAILED for cleanup & alerts.

5 · Run-if — conditional execution on outcomes

`run_if` value	Runs when…	Use for
`ALL_SUCCESS` (default)	every dependency succeeded	normal flow
`AT_LEAST_ONE_SUCCESS`	≥1 dependency succeeded	best-effort merge
`NONE_FAILED`	none failed (skips allowed)	tolerant continue
`ALL_DONE`	all finished (success or fail)	cleanup
`AT_LEAST_ONE_FAILED`	≥1 dependency failed	alerting
`ALL_FAILED`	all failed	fallback path

  - task_key: alert_on_failure
    depends_on: [{ task_key: ingest_all }, { task_key: publish_gold }]
    run_if: AT_LEAST_ONE_FAILED          # fire only when something upstream failed
    notebook_task: { notebook_path: ../src/alert.py }

Why it matters

Production pipelines need branching, looping, and recovery — not a fixed sequence.
"Process N tables in one job?" → for-each; "skip publish if validation failed?" → if/else + run-if — common interview probes.

🔁 Real-world use case: a job uses for-each to ingest 12 tables in parallel, sets a row-count task value, an if/else publishes to gold only if counts look sane, and a run-if AT_LEAST_ONE_FAILED task fires an alert.

Uses, edge cases & limitations

Uses: data-driven branching, fan-out over many inputs, passing counts/flags, conditional alerting/cleanup.
Edge cases: task values are small — for big data write to a table and pass a pointer; for-each parallelism is bounded (concurrency + limits), batch huge lists; run_if defaults to ALL_SUCCESS.
Limitation: control flow is job/task-level — fine-grained row logic belongs in task code/SDP; over-using if/else hurts readability (sometimes separate jobs are clearer).

Common gotchas

❌ Stuffing large data into a task value — it's for small values; use a table.
❌ Forgetting run_if defaults to ALL_SUCCESS — cleanup/alert won't run after a failure unless set to AT_LEAST_ONE_FAILED / ALL_DONE.
❌ A notebook if in code when you need task-level branching (so DAG/monitoring reflects it) — use condition_task.
❌ Unbounded for-each over thousands of items without batching / concurrency.