For Each Loops in Lakeflow Jobs

1.A For each task wraps exactly one nested task

The For each task runs another task in a loop and passes a different set of parameters to each iteration. A typical use is applying the same transformations to many tables, with one table name from a list going to each iteration.

A For each task is always made of two tasks. The For each task itself holds the list of inputs, and the nested task is the work that runs once per input. The nested task can be any standard Lakeflow Jobs task type, such as a notebook or a SQL task, with the usual configuration options. The one exception: it can't be another For each task. In the Jobs UI, the nested task appears as a node inside the For each node.

Iterations that don't depend on each other can run concurrently. In the UI, the Concurrency field sets how many iterations may run in parallel, and it defaults to 1. If you leave it at 1, a list of fifty tables runs one table at a time.

Sources1

2.Feeding the loop: inputs and their size limits

The For each task's Inputs field is a JSON array, and each element goes to one iteration. Elements can be strings, numbers, booleans, key-value pairs, or arbitrarily complex JSON objects. The nested task reads its element through its own parameters:

- {{input}} is the whole array element for this iteration, such as a table name. - {{input.<key>}} is one field of an object element. With the input [{"table": "sales"}, {"table": "orders"}], the nested task reads {{input.table}}.

You don't have to type the array by hand. Inputs can come from a JSON literal, a task value set by an earlier task, or a job parameter. Each source has a different size limit, and exam questions like to test these.

The 5,000-character limit applies to the text in the Inputs box. A short reference such as {{tasks.generate_countries_list.values.countries}} can expand to a value of up to 48 KB, which is why task values work well for long, computed lists. The producing task builds the list in Python and publishes it with dbutils.jobs.taskValues.set(key = "countries", value = countries_array). A SQL task can feed the loop as well: the control-table tutorial passes {{tasks.read_segments.output.rows}}, the SQL task's rows as a JSON array of objects.

In a bundle, the same structure is a for_each_task with a required inputs string, a required nested task map, and an optional concurrency.

resources:
  jobs:
    my_job:
      name: my_job
      tasks:
        - task_key: generate_countries_list
          notebook_task:
            notebook_path: ../src/generate_countries_list.ipnyb
        - task_key: process_countries
          depends_on:
            - task_key: generate_countries_list
          for_each_task:
            inputs: '{{tasks.generate_countries_list.values.countries}}'
            task:
              task_key: process_countries_iteration
              notebook_task:
                notebook_path: ../src/process_countries_notebook.ipnyb

Sources12

3.After the loop: dependencies and iteration outputs

The For each task is the top-level task, and downstream tasks depend on it, not on the nested task. Each iteration can still call dbutils.jobs.taskValues.set() like any other task, and a downstream task can collect all of those values.

The key to remember is that you read them through the nested task's key, not the For each task's. That key returns one list ordered by iteration index. Through dbutils.jobs.taskValues.get() you get a Python list, and the dynamic value reference {{tasks.process.values.result}} resolves to the same array. If an iteration didn't set the key, its slot is kept: None from get(), or null in a reference. So a missing value in the second of three iterations gives [1, None, 3].

There are limits. The assembled array can't exceed about 48 KB, and a single parameter value can contain at most three aggregated task-value references. This feature is in Beta, is on by default, and requires Databricks Runtime 15.4 LTS or above.

results = dbutils.jobs.taskValues.get(taskKey =  ? , key = "result")
# results == [1, 2, 3]

You run and manage these jobs the same way as any other job. The only difference is in run history: a For each task's run history shows as a table of iterations. Like the If/else task, a For each task fails if the upstream task that provides its input values is disabled.

Sources31

4.Scaling the pattern: lookup files and control tables

The size limits still apply when you need to pass more than 48 KB of parameters. The documented workaround is to put the full configuration in a JSON file and pass each iteration only a lookup key. In the example, the For each Inputs is ["table_1","table_2","table_3"]. The nested notebook has a parameter key set to {{input}}, and it uses that key to find the step's arguments in the file. The steps don't depend on each other, so concurrency can be set above 1.

# look up step and arguments
key = dbutils.widgets.get("key")
current_step = next((step for step in config['steps'] if step['key'] == key), None)
if current_step is None:
    raise ValueError(f"Could not find step '{key}' in the configuration")

args = SimpleNamespace(**current_step["args"])

A related problem is a list that changes often. If the list is hard-coded in the job, every change means editing and redeploying. The control-table pattern keeps the list in a Delta table instead. A SQL task reads the table, and its {{tasks.read_segments.output.rows}} output becomes the For each inputs. The nested task then gets each row's columns as {{input.property_type}} and {{input.min_price}}. To add or remove work, you change a row, and the next run picks up the change. For the nested task, use a notebook when the logic needs procedural code or libraries. Use a SQL task when a single query is enough, and note that a SQL task needs a SQL warehouse.

Sources45

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.You can nest a For each task inside another For each task to loop over two dimensions.Why is that wrong?

   The nested task must be a standard task type. Another For each task is not allowed.

   Covered in A For each task wraps exactly one nested task

2. 2.To read every iteration's task value, you use the For each task's name as the taskKey.Why is that wrong?

   You use the nested task's key, which returns all iterations' values as one ordered list.

   Covered in After the loop: dependencies and iteration outputs

Practise it for real

Build a metadata-driven job in which a SQL task reads a control table and a For each task runs one analysis per row

1. 1.In the SQL editor, run USE CATALOG <catalog-name>; CREATE SCHEMA IF NOT EXISTS config; then CREATE OR REPLACE TABLE config.property_segments with the rows ('Urban Year-Round', 150), ('Summer Getaway', 200) and ('Ski Resort', 250) AS t(property_type, min_price).

   Why: The control table, not the job definition, decides which segments get processed.

   You should see: config.property_segments contains three rows, one per segment.

2. 2.Create a notebook named run_segment_analysis. It should call dbutils.widgets.text() for property_type and min_price before calling dbutils.widgets.get(), then query samples.wanderbricks.properties filtered by those parameters.

   Why: Default widget values let you test the notebook on its own, and the job overrides them during each iteration.

   You should see: Run interactively, it returns a count and average price for Ski Resort listings priced at 250 or more.

3. 3.Create a job with a SQL task named read_segments that selects from config.property_segments.

   Why: The SQL task's output rows become the list the loop iterates over.

   You should see: The task's output is a JSON array of row objects.

4. 4.Add a For each task named process_segments that depends on read_segments, with Inputs set to {{tasks.read_segments.output.rows}}. Set concurrency above 1.

   Why: Each row goes to one iteration, and the segments are independent, so they can run in parallel.

   You should see: The For each node shows an empty slot for a nested task.

5. 5.Add the run_segment_analysis notebook as the nested task, with parameters property_type = {{input.property_type}} and min_price = {{input.min_price}}. Then run the job.

   Why: {{input.<key>}} passes each row's fields to its iteration.

   You should see: The run history for the For each task lists three iterations, one per segment.