Skip to main content
Last updated on

Use a control table to drive a For each job

When you run the same processing across many inputs, such as markets, source tables, customers, or date partitions, hardcoding that list in your job means editing code and redeploying every time the list changes. Instead, store the list in a control table that the job reads at run time. To add or remove work, you update a row in the table, and the next job run picks up the change without edits to the job itself. This is a metadata-driven pattern: the data, not the code, controls what the job processes.

This tutorial builds a job that uses this pattern on the preinstalled Wanderbricks sample dataset, so you can run it end to end without creating any source data. The scenario is a vacation-rental platform that runs the same price analysis for each property segment (such as Ski Resort or Urban Year-Round). A control table lists the segments to analyze, a SQL task reads that table, and a For each task runs the analysis one time per segment, in parallel.

How it works​

The job wires three tasks together in sequence:

Task

Type

What it does

read_segments

SQL

Reads the control table and captures the rows as a JSON array

process_segments

For each

Iterates over the row array, launching the nested task one time per row

run_segment_analysis

Notebook or SQL (nested inside For each)

Runs one time per row, using that row's values to analyze one property segment

Task

Type

What it does

read_segments

SQL

Reads the control table and captures the rows as a JSON array

process_segments

For each

Iterates over the row array, launching the nested task one time per row

run_segment_analysis

Notebook or SQL (nested inside For each)

Runs one time per row, using that row's values to analyze one property segment

The flow is read_segments → process_segments → run_segment_analysis (one time per row). The SQL task's output, a JSON array of row objects, flows into the For each task's Inputs field through the dynamic value reference {{tasks.read_segments.output.rows}}. The For each task then passes each row's fields to the nested task as parameters, available as {{input.property_type}} and {{input.min_price}}.

Prerequisites​

  • A Databricks workspace with permission to create jobs and notebooks.
  • Permission to create tables in Unity Catalog, and permission to create a schema in a catalog (the USE CATALOG and CREATE SCHEMA privileges) to hold the control table.
  • A SQL warehouse to run the SQL tasks. If you do not have one, see Create a SQL warehouse.
  • The samples catalog, which is available in every Unity Catalog-enabled workspace. The tutorial reads from samples.wanderbricks.properties, so there is no source data to set up.

Step 1: Create the control table​

The control table is the source of truth for the list of segments your job processes. To change what the job does, you update this table, not the job.

Run the following SQL in a Databricks notebook or the SQL editor. The first statement creates a schema to hold the control table, and the second creates the table with one row per property segment and the minimum listing price to include in that segment's analysis:

SQL
USE CATALOG <catalog-name>;

CREATE SCHEMA IF NOT EXISTS config;

CREATE OR REPLACE TABLE config.property_segments AS
SELECT * FROM VALUES
  ('Urban Year-Round', 150),
  ('Summer Getaway', 200),
  ('Ski Resort', 250)
AS t(property_type, min_price);

Replace <catalog-name> with a catalog you can create schemas in, such as your workspace catalog. Use the same catalog everywhere the tutorial references config.property_segments, including the lookup query in Step 3.

After this step, config.property_segments contains three rows, one per segment. Each row carries the two values the job passes to each iteration: the property_type to analyze and the min_price floor to filter on.

Step 2: Write the analysis logic​

The nested task inside the For each task runs one time per row of the control table, receiving that row's property_type and min_price as parameters. You can write this logic as a notebook task or a SQL task. Choose based on your business logic:

  • Use a notebook task when the per-iteration logic needs procedural code, multiple languages, or libraries (for example, a data science or machine learning step).
  • Use a SQL task when the logic is a single query or transformation that you can express declaratively. A SQL task needs a SQL warehouse.

Both variants below produce the same result: for the segment being processed, the number of listings at or above its price floor and their average price.

Create a new notebook at a path such as /Workspace/Users/<username>/run_segment_analysis. This notebook runs one time per iteration of the For each task, receiving a different segment each time.

Add the following code to the notebook:

Python
# Set default values so you can run the notebook on its own while developing.
# When the notebook runs inside a For each task, the job overrides these defaults.
dbutils.widgets.text("property_type", "Ski Resort", "Property type")
dbutils.widgets.text("min_price", "250", "Minimum price")

# Read the parameters passed by the For each task.
property_type = dbutils.widgets.get("property_type")
min_price = dbutils.widgets.get("min_price")

result = spark.sql(
    """
    SELECT :property_type AS property_type,
           COUNT(*) AS property_count,
           ROUND(AVG(base_price), 2) AS avg_price
    FROM samples.wanderbricks.properties
    WHERE property_type = :property_type
      AND base_price >= :min_price
    """,
    args={"property_type": property_type, "min_price": min_price},
)
display(result)
note

Call dbutils.widgets.text() before dbutils.widgets.get(). If you call get first, running the notebook outside a job raises an InputWidgetNotDefined error.

Step 3: Create the lookup query​

The lookup task reads the control table through a saved query. As in Step 2, create and save the query in the SQL editor now, then attach it to the lookup task in Step 4.

  1. In your Databricks workspace, click Plus icon. New > Query icon. Query to open the SQL editor.

  2. Enter the following, using the same catalog you chose in Step 1:

    SQL
    SELECT property_type, min_price FROM <catalog-name>.config.property_segments;

    The name is fully qualified because the SQL warehouse that runs this query might default to a different catalog than the one you created the table in.

  3. Click the title New Query <date> in the tab heading of your SQL file, and give it the name read_segments. Then click Save to move it to a folder where you want to store it.

Step 4: Create and configure the job​

With both queries saved, create the job and add its two tasks: the SQL lookup task that reads the control table, and the For each task that runs the analysis for each row.

Create the job​

In your Databricks workspace, in the sidebar click Plus icon. New > Workflows icon. Job. Give the job a descriptive name, such as Segment Analysis.

Configure the SQL lookup task​

This task reads the control table and makes its rows available to the For each task by running the read_segments query you saved in Step 3.

  1. Click the SQL query tile to configure the first task. If the SQL query tile is not available, click Add another task type and search for SQL query.
  2. Set Task name to read_segments.
  3. If necessary, select SQL query from the Type drop-down menu.
  4. In the SQL query field, select the read_segments query you saved in Step 3.
  5. Set SQL warehouse to a warehouse in your workspace.
  6. Click Create task.

When this task runs, Databricks captures the result as a JSON array in tasks.read_segments.output.rows. SQL task output is always returned as a JSON array, so you do not need any extra configuration. The general form of the reference is tasks.<task-name>.output.rows, where <task-name> matches the task name you set. The output looks like this:

JSON
[
  { "property_type": "Urban Year-Round", "min_price": 150 },
  { "property_type": "Summer Getaway", "min_price": 200 },
  { "property_type": "Ski Resort", "min_price": 250 }
]

Configure the For each task​

The For each task reads the SQL output and launches one nested task run per row.

  1. Click Plus icon. Add task and select For each.

  2. Set Task name to process_segments.

  3. Verify that Depends on is set to read_segments.

  4. In the Inputs field, enter the row array captured by the SQL task:

    {{tasks.read_segments.output.rows}}

  5. Set Concurrency to 2 to run two iterations in parallel. Increase this value when your nested task supports higher parallelism.

  6. To complete this task, click Add a task to loop over and configure the nested task that runs on each iteration.

The For each task and its nested task are created together as a single task. Configure the nested task based on the type you chose in Step 2:

  1. Set Task name to run_segment_analysis.

  2. Set Type to Notebook.

  3. Set Path to the notebook you created in Step 2.

  4. Click Parameters, then click Add to add each parameter:

    • Key: property_type, Value: {{input.property_type}}
    • Key: min_price, Value: {{input.min_price}}

    Each {{input.<key>}} reference resolves to the matching field from the current iteration's row.

  5. Click Create task to create the For each task and its nested task together.

Your job Directed Acyclic Graph (DAG) now shows read_segments flowing into process_segments, with the nested task inside the For each node.

Step 5: Run the job and verify​

  1. Click Run now to trigger the job.
  2. Select the Runs tab to see the run. The first run of a job takes a few minutes to start compute; when it completes, it appears in the list.
  3. Click the process_segments node to expand the For each task.
  4. The run page shows a table of iterations, one row per segment, each with its status, start time, and duration.
  5. Click any iteration row to open its output and confirm it analyzed the expected segment.

You can see the results of each iteration independently. If a specific iteration fails, you can rerun only that iteration from the job run page without rerunning the whole job.

Extend the pattern​

To add a segment to the analysis, insert a row into the control table:

SQL
INSERT INTO <catalog-name>.config.property_segments VALUES ('Historical Place', 100);

The next job run includes the new segment, with no job configuration changes or notebook edits.

This same pattern works for any case where you want data to drive iteration:

  • Per-customer processing: One row per customer ID. The nested task applies customer-specific transformations or delivers to customer-specific destinations.
  • Table ingestion: One row per source table name. The nested task reads and ingests each table.
  • Backfill processing: One row per date partition. The nested task reprocesses historical data for that partition.
  • Feature flag-driven execution: One row per enabled feature or experiment. The nested task activates the corresponding logic.

To stop processing a row without deleting it, add your own column to the control table (such as an active flag) and filter on it in the SQL lookup task. This is an ordinary column that you define and populate; the For each task has no built-in concept of it. First add the column, then set the existing rows to TRUE:

SQL
ALTER TABLE <catalog-name>.config.property_segments ADD COLUMN active BOOLEAN;
UPDATE <catalog-name>.config.property_segments SET active = TRUE;

Then filter on it in the read_segments query so only active rows drive iteration:

SQL
SELECT property_type, min_price FROM <catalog-name>.config.property_segments WHERE active = TRUE;

Additional resources​

On this page