Skip to main content
Last updated on

Copy multiple tables incrementally with a For each task

When you need to copy data from many source tables into Unity Catalog tables on a schedule, copying all rows on every run is slow and expensive. Use a watermark to track the last-processed row for each table and copy only new rows on each run.

This tutorial shows you how to build a metadata-driven job that:

  • Stores the list of source tables and their watermark state in a Delta control table
  • Uses a For each task to process each table in parallel
  • Copies only rows added since the last successful run
  • Updates the watermark after each successful copy

How it works​

The job uses three task types wired together in sequence:

Task

Type

What it does

read_watermarks

SQL

Reads the watermark control table and returns one row per source table

copy_tables

For each

Iterates over {{tasks.read_watermarks.output.rows}}, running the nested task once per source table

copy_incremental (nested)

Notebook

Reads rows added since the last watermark, writes them to the target table, and advances the watermark

Task

Type

What it does

read_watermarks

SQL

Reads the watermark control table and returns one row per source table

copy_tables

For each

Iterates over {{tasks.read_watermarks.output.rows}}, running the nested task once per source table

copy_incremental (nested)

Notebook

Reads rows added since the last watermark, writes them to the target table, and advances the watermark

The SQL task output—a JSON array of row objects—flows into the For each task's Inputs field using {{tasks.read_watermarks.output.rows}}. The nested notebook receives source_table, target_table, watermark_column, and last_watermark for each iteration.

Prerequisites​

  • A Databricks workspace with permission to create jobs and notebooks
  • Permission to create schemas and tables in Unity Catalog
  • A SQL warehouse to run SQL tasks

This tutorial reads from the samples.wanderbricks sample dataset and writes to a schema named by the catalog and schema variables at the top of each code block. Those variables default to main.example_output. To write somewhere else, change both values consistently across every block. The example creates the schema if it doesn't exist.

Step 1: Create the watermark control table​

The watermark control table is the source of truth for which tables to process and how far each table has been copied. Each row represents one source table.

Run the following SQL to create the control table and register two source tables. The example registers samples.wanderbricks.users and samples.wanderbricks.properties, copying each into a target table in your own schema. The catalog and schema variables set both the control table's location and the target table names:

SQL
DECLARE OR REPLACE VARIABLE catalog STRING DEFAULT 'main';
DECLARE OR REPLACE VARIABLE schema  STRING DEFAULT 'example_output';

CREATE SCHEMA IF NOT EXISTS IDENTIFIER(catalog || '.' || schema);

CREATE OR REPLACE TABLE IDENTIFIER(catalog || '.' || schema || '.watermarks') (
  source_table      STRING NOT NULL,
  target_table      STRING NOT NULL,
  watermark_column  STRING NOT NULL,
  last_watermark    TIMESTAMP NOT NULL
);

INSERT INTO IDENTIFIER(catalog || '.' || schema || '.watermarks') VALUES
  ('samples.wanderbricks.users',      catalog || '.' || schema || '.users',      'created_at', '1970-01-01'),
  ('samples.wanderbricks.properties', catalog || '.' || schema || '.properties', 'created_at', '1970-01-01');

Both source tables use created_at as the watermark column. This is an insert-time timestamp that only ever increases as new rows arrive. Setting last_watermark to 1970-01-01 on the first run causes the notebook to copy all existing rows. This acts as an initial full load. Subsequent runs copy only rows added after the previous run.

note

Recreating the control table resets each last_watermark to 1970-01-01, but it does not clear the target tables. If you rerun this step and then run the job again, the notebook treats both sources as never-copied and appends every historical row a second time. To start over cleanly, drop the target tables when you recreate the control table.

Step 2: Write the copy notebook​

The notebook runs once per table iteration. It reads the watermark, filters the source, writes to the target, and advances the watermark.

Create a notebook at a path such as /Workspace/Users/<username>/copy_incremental and add the following code. Widget defaults let you run and test the notebook directly. When it runs inside the For each task, the job overrides them with each iteration's values, including the catalog and schema that locate the control table.

The code reads only the rows added since the last watermark and appends them to the target table, creating it if it doesn't exist. It then computes the high water mark of the rows it just wrote and advances the control table so the next run starts from there:

Python
from pyspark.sql.functions import max as spark_max

# Widget defaults let you run the notebook directly; the For each task overrides them per iteration
dbutils.widgets.text("catalog",          "main",                            "Catalog")
dbutils.widgets.text("schema",           "example_output",                  "Schema")
dbutils.widgets.text("source_table",     "samples.wanderbricks.users",      "Source table")
dbutils.widgets.text("target_table",     "main.example_output.users",       "Target table")
dbutils.widgets.text("watermark_column", "created_at",                      "Watermark column")
dbutils.widgets.text("last_watermark",   "1970-01-01",                      "Last watermark")

catalog          = dbutils.widgets.get("catalog")
schema           = dbutils.widgets.get("schema")
source_table     = dbutils.widgets.get("source_table")
target_table     = dbutils.widgets.get("target_table")
watermark_column = dbutils.widgets.get("watermark_column")
last_watermark   = dbutils.widgets.get("last_watermark")

# Read only rows newer than the last watermark. A strict > can skip rows that share the
# stored high-water timestamp; for insert-only sources with distinct timestamps this is safe.
new_rows = spark.table(source_table).filter(f"{watermark_column} > '{last_watermark}'")

row_count = new_rows.count()
print(f"Copying {row_count} new rows from {source_table}")

if row_count > 0:
    # Append the new rows, creating the target table on the first run
    new_rows.write.format("delta").mode("append").saveAsTable(target_table)

    # Compute the high-water mark from the rows just written
    new_watermark = new_rows.agg(spark_max(watermark_column)).collect()[0][0]

    # Advance the control table so the next run starts from here
    spark.sql(f"""
        UPDATE {catalog}.{schema}.watermarks
        SET last_watermark = CAST('{new_watermark}' AS TIMESTAMP)
        WHERE source_table = '{source_table}'
    """)

    print(f"Watermark for {source_table} advanced to {new_watermark}")
else:
    print(f"No new rows for {source_table}, watermark unchanged")
note

This notebook uses append mode, which is suitable when the source contains only inserts, as samples.wanderbricks.users and samples.wanderbricks.properties do. If your source contains updates, watermark on the update timestamp and use a MERGE statement instead of write.mode("append") to upsert rows into the target table. See Upsert into a Delta Lake table using merge for merge syntax.

Step 3: Create the job​

In your Databricks workspace, click Workflows in the sidebar, then click Create job. Give the job a name such as Incremental table copy.

Step 4: Configure the watermark lookup task​

The SQL task reads the control table and makes the result available to the For each task. Because the query declares the catalog and schema variables that locate the control table, it must run as a multi-statement SQL file rather than in the task's inline SQL field.

  1. Create a SQL file in your workspace, such as /Workspace/Users/<username>/read_watermarks.sql, with the following content. Set catalog and schema to the same values you used in Step 1:

    SQL
    DECLARE OR REPLACE VARIABLE catalog STRING DEFAULT 'main';
    DECLARE OR REPLACE VARIABLE schema  STRING DEFAULT 'example_output';

    SELECT source_table, target_table, watermark_column, last_watermark
    FROM IDENTIFIER(catalog || '.' || schema || '.watermarks');

  2. In the job, click Add task.

  3. Set Task name to read_watermarks.

  4. Set Type to SQL, then set SQL task to File.

  5. Set Path to the SQL file you created.

  6. Set SQL warehouse to a warehouse in your workspace.

  7. Click Create task.

When this task runs, Databricks captures the result as a JSON array in tasks.read_watermarks.output.rows. After an initial full load, each last_watermark reflects the most recent row copied from that source:

JSON
[
  {
    "source_table": "samples.wanderbricks.users",
    "target_table": "main.example_output.users",
    "watermark_column": "created_at",
    "last_watermark": "2025-07-30T23:05:18.000Z"
  },
  {
    "source_table": "samples.wanderbricks.properties",
    "target_table": "main.example_output.properties",
    "watermark_column": "created_at",
    "last_watermark": "2025-07-30T00:00:00.000Z"
  }
]

Step 5: Configure the For each task​

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

  1. Click Add task and set Depends on to read_watermarks.

  2. Set Task name to copy_tables.

  3. Set Type to For each.

  4. In the Inputs field, enter:

    {{tasks.read_watermarks.output.rows}}

  5. Set Concurrency to 2 to copy two tables at a time. Increase this value if your warehouse can support higher parallelism.

  6. Click Add a task to loop over to configure the nested task.

  7. Set Task name to copy_incremental.

  8. Set Type to Notebook.

  9. Set Path to the path of the notebook you created in Step 2.

  10. Click Parameters, then click Add to add each of the following parameters:

    Key

    Value

    catalog

    main

    schema

    example_output

    source_table

    {{input.source_table}}

    target_table

    {{input.target_table}}

    watermark_column

    {{input.watermark_column}}

    last_watermark

    {{input.last_watermark}}

    Key

    Value

    catalog

    main

    schema

    example_output

    source_table

    {{input.source_table}}

    target_table

    {{input.target_table}}

    watermark_column

    {{input.watermark_column}}

    last_watermark

    {{input.last_watermark}}

    Set catalog and schema to the same values you used in Step 1 so the notebook advances the control table the SQL task reads. Each {{input.<key>}} reference resolves to the corresponding field from the current iteration's row.

  11. Click Create task.

Step 6: Run the job and verify​

  1. Click Run now to trigger the job.
  2. On the job run page, click the copy_tables node to expand the For each task.
  3. The run page shows a table of iterations—one row per source table—each displaying its status, start time, and duration.
  4. Click any iteration to view the notebook output and confirm the row count and watermark update.

To confirm the watermark advanced, run the following query after the job completes. Set catalog and schema to the same values you used in the previous steps:

SQL
DECLARE OR REPLACE VARIABLE catalog STRING DEFAULT 'main';
DECLARE OR REPLACE VARIABLE schema  STRING DEFAULT 'example_output';

SELECT source_table, last_watermark
FROM IDENTIFIER(catalog || '.' || schema || '.watermarks');

Each last_watermark value should now reflect the timestamp of the most recently copied row. If a value is still 1970-01-01, the source table contained no rows matching the filter, or the copy task encountered an error — check the task run output for details.

Extend the pattern​

Each snippet declares the same catalog and schema variables used in the previous steps. Set them to the values that locate your control table.

Add a new source table: Insert a row into the control table. The next job run picks it up automatically, starting with a full load from 1970-01-01. This snippet assumes you have already added the active column from Pause a table below, so it sets active to true. The two extensions can be applied in either order; if you have not added the column yet, drop the last value and its column:

SQL
DECLARE OR REPLACE VARIABLE catalog STRING DEFAULT 'main';
DECLARE OR REPLACE VARIABLE schema  STRING DEFAULT 'example_output';

INSERT INTO IDENTIFIER(catalog || '.' || schema || '.watermarks')
  (source_table, target_table, watermark_column, last_watermark, active)
VALUES
  ('samples.wanderbricks.hosts', catalog || '.' || schema || '.hosts', 'joined_at', '1970-01-01', TRUE);

Pause a table: Add an active column, backfill it to true for existing rows, then filter on it in the SQL file task. Delta requires adding the column and setting its value in separate statements:

SQL
DECLARE OR REPLACE VARIABLE catalog STRING DEFAULT 'main';
DECLARE OR REPLACE VARIABLE schema  STRING DEFAULT 'example_output';

ALTER TABLE IDENTIFIER(catalog || '.' || schema || '.watermarks') ADD COLUMN active BOOLEAN;

UPDATE IDENTIFIER(catalog || '.' || schema || '.watermarks') SET active = TRUE;

Then add WHERE active = TRUE to the SELECT in your read_watermarks.sql file so the job skips paused tables.

Backfill a table: Reset its watermark to re-copy from a specific point:

SQL
DECLARE OR REPLACE VARIABLE catalog STRING DEFAULT 'main';
DECLARE OR REPLACE VARIABLE schema  STRING DEFAULT 'example_output';

UPDATE IDENTIFIER(catalog || '.' || schema || '.watermarks')
SET last_watermark = '2025-01-01'
WHERE source_table = 'samples.wanderbricks.users';

Additional resources​

On this page