Building the Curated Pipeline: Orchestrating Multi-Level Notebook Execution in Microsoft Fabric

Why a Two-Pipeline Architecture?

At first glance, you might ask: why not combine landing and curation into a single pipeline? The answer lies in separation of concerns and operational flexibility. The landing pipeline is purely mechanical: it reads config, pulls data from SQL Server, archives old versions, and updates watermarks. It's simple, fast, and rarely changes.

The curated pipeline is different. It contains business logic, complex transformations, joins across multiple tables, and orchestrates 3-5 interdependent notebooks. If something breaks in curation (logic error, schema mismatch, data quality issue), you want to fix it and re-run without re-ingesting all your source data. Two pipelines give you that independence.

Moreover, the landing pipeline can run on a tight schedule (every 15 minutes). The curated pipeline might only run hourly or nightly because transformations are heavier. Separating them lets you tune SLAs independently.

The Invoke Pipeline Activity — Always Fresh Landing Data

The curated pipeline (pl_ingest_curated) starts with a single activity: invoke_pl_ingest_landing. This is an Invoke Pipeline activity that calls the landing pipeline we built in Part 5.

Why invoke the landing pipeline first? Because curation depends on fresh landing data. By invoking it at the start, you ensure that before any transformation logic runs, the landing layer is up-to-date. If new drivers were added to SQL Server in the last 15 minutes, they're now in cab_landing.DRIVERSENTITY. Only then does curation proceed.

The Invoke activity is configured to:

• Pipeline: pl_ingest_landing
• Wait on completion: Checked (the curated pipeline waits for landing to finish before proceeding)
• Parameters: None (the landing pipeline has no parameters)

The Lookup — Reading Notebook Metadata from Config

Once the landing pipeline completes, lkp_config_curated executes. This Lookup activity reads the notebook configuration table:

SELECT * FROM config.curated_control_tbl WHERE enabled = 1

This returns rows like:

Each notebook has a level (1 or 2) and an optional dependent_notebook field. Level 1 notebooks are independent—they read from landing and write to transient and curated tables. Level 2 notebooks depend on Level 1 notebooks—they join data from multiple transient layers.

Level 1 vs Level 2 — The Dependency Hierarchy

Understanding the two-level architecture is critical. Here's what each level does:

Level 1 Notebooks: Independent Transformers

A Level 1 notebook like nb_com_drivers_d has a simple job:

1. Read from cab_landing.DRIVERSENTITY
2. Apply business logic: de-duplication, cleansing, type conversions, calculated columns
3. Write to transient.com_drivers_d (staging layer for other notebooks to use)
4. Also write to curated.com_drivers_d (dimension table for reporting)
5. Write to warehouse.com_drivers_d (for Power BI semantic model)

The Level 1 notebook doesn't depend on any other notebook. If it finds 0 new rows in landing, it exits cleanly.

Level 2 Notebooks: Dependent Transformers

A Level 2 notebook like nb_cab_trips_f is more complex:

1. Read from transient.com_trips_header_f and transient.com_trips_line_f (outputs of Level 1 notebooks)
2. Join on trip ID: fact table = header + line detail
3. Write to transient.cab_trips_f
4. Also write to curated.cab_trips_f (fact table)
5. Write to warehouse.cab_trips_f

The Level 2 notebook depends on the outputs of Level 1. It can't run until Level 1 completes.

Filter Activity — Splitting the Execution into Levels

After the Lookup, the pipeline encounters its first Filter activity: filter_level_1_nb. This activity filters the lookup output to only rows where notebook_level = 1:

@equals(item().notebook_level, 1)

The output of this filter is passed to the first ForEach loop. Only Level 1 notebooks enter this loop.

ForEach1 — Running Level 1 Notebooks in Parallel

ForEach1 iterates over the filtered Level 1 notebooks. Like the landing pipeline's ForEach, parallelism is enabled (Sequential Run is unchecked). This means all 3 Level 1 notebooks run concurrently.

Inside ForEach1: The Notebook Activity

The notebook activity run_nb_curated_level_1 is configured to:

• Workspace: Your Fabric workspace
• Notebook: Retrieved dynamically: @item().notebook_name
• Parameters: nb_id = @item().nb_id

The notebook receives the nb_id (the Fabric GUID for the notebook) as a parameter. Inside the notebook, code looks like:

"""Retrieve notebook ID from pipeline parameter"""
nb_id = dbutils.widgets.get("nb_id")

"""Query landing table based on notebook's assigned table"""
SELECT * FROM cab_landing.DRIVERSENTITY
WHERE modified_date >= '2026-04-08 12:30:35'
"""The notebook processes data from the landing schema, while incremental filtering is handled upstream in the landing pipeline"""

The notebook reads landing data, applies transformations, and writes to transient and curated. If it finds 0 rows, it can exit early with a no-op.

Inside ForEach1: The Watermark Stored Procedure

After the notebook completes successfully, a stored procedure sp_update_curated_config_watermark runs:

EXEC config.update_curated_config_watermark
    @notebook_name = @item().notebook_name,
    @last_runtime = GETUTCDATE()

This updates the last_pipeline_runtime column for this Level 1 notebook. This watermark is important: Level 2 notebooks will read it to know which records to include when joining.

Filter Level 2 — Ensuring Sequential Level Execution

After ForEach1 completes (all 3 Level 1 notebooks finish), the pipeline encounters filter_level_2_nb. This filters the Lookup output to only Level 2 notebooks.

This filter happens after ForEach1 finishes. This is the orchestration trick: by placing the filter and ForEach2 sequentially (not in parallel), we enforce that Level 2 always runs after Level 1. If you put both ForEach loops in parallel, Level 1 and Level 2 might run simultaneously, causing Level 2 to read stale transient data.

ForEach2 — Running Complex Notebooks

ForEach2 iterates over Level 2 notebooks in parallel. Inside, the notebook activity run_nb_curated_level_2 runs the Level 2 notebook with the same nb_id parameter.

The key difference: Level 2 notebooks read the watermark of their dependent notebooks (not their own) from the config table. For example, nb_cab_trips_f depends on nb_com_trips_header_f and nb_com_trips_line_f. Inside the notebook, you'll see code that reads:

"""Get the watermark of dependent notebooks to know what's new"""
header_runtime = spark.sql("""
    SELECT last_pipeline_runtime
    FROM config.curated_control_tbl
    WHERE notebook_name = 'nb_com_trips_header_f'
""").collect()[0][0]

"""Now filter transient data to only records from that point forward"""
new_trips = spark.sql(f"""
    SELECT *
    FROM transient.com_trips_header_f
    WHERE modified_date >= CAST('{header_runtime}' AS DATETIME)
""")

How Level 2 Avoids Stale Transient Data

This is the critical insight. Even if a Level 1 notebook runs with no new data, its watermark value is still available for downstream processing. When a Level 2 notebook starts, it reads the watermark datetime from the Level 1 notebook and applies it as a filter on the transient layer data. Since there is no new data beyond that watermark, the filter returns 0 rows. As a result, the Level 2 notebook exits gracefully without processing any records, thereby preventing duplicate data from being created.

The pattern is: Level 2 notebooks always read the watermark of their dependencies, not their own watermark.

The Full Execution Timeline (Visual)

Here's what a real execution looks like:

1. 9:00:00 AM: pl_ingest_curated starts.
2. 9:00:01 AM: Invoke pl_ingest_landing. Landing pipeline begins.
3. 9:00:30 AM: Landing pipeline completes. 150 new drivers, 200 new trips_header, 500 new trips_line records.
4. 9:00:31 AM: Lookup returns 4 rows (3 Level 1 + 1 Level 2).
5. 9:00:32 AM: Filter Level 1 passes 3 rows to ForEach1.
6. 9:00:33 AM: ForEach1 starts all 3 Level 1 notebooks in parallel:
   • nb_com_drivers_d reads 150 drivers from landing, deduplicates (10 dupes), writes 140 unique drivers to transient and curated. Updates watermark to 9:00:45.
   • nb_com_trips_header_f reads 200 header rows, applies business logic, writes to transient and curated. Updates watermark to 9:00:50.
   • nb_com_trips_line_f reads 500 line rows, aggregates, writes to transient and curated. Updates watermark to 9:00:48.
7. 9:00:51 AM: All Level 1 notebooks finish. ForEach1 exits.
8. 9:00:52 AM: Filter Level 2 passes 1 row (cab_trips_f) to ForEach2.
9. 9:00:53 AM: ForEach2 starts nb_cab_trips_f:
   • Reads last_pipeline_runtime of nb_com_trips_header_f (9:00:50) and nb_com_trips_line_f (9:00:48).
   • Filters transient.com_trips_header_f WHERE modified_date >= 9:00:48 (gets 200 rows).
   • Filters transient.com_trips_line_f WHERE modified_date >= 9:00:48 (gets 500 rows).
   • Joins on trip_id, creates fact table, writes to transient.cab_trips_f and curated.cab_trips_f.
   • Updates its own watermark to 9:01:10.
10. 9:01:11 AM: ForEach2 completes. pl_ingest_curated finishes.

Testing the Pipeline

Testing a multi-level orchestration requires checking both logic and orchestration:

Orchestration Tests

• Level Sequencing: Add a test row to landing. Run pl_ingest_curated. Check that Level 1 completes before Level 2 starts (look at activity timestamps in the pipeline run history).
• No Early Execution: Disable one Level 1 notebook. Run the pipeline. Verify that the other Level 1 notebooks still run in parallel, but Level 2 waits for all Level 1 to complete.
• Filter Correctness: Run the pipeline. Check the filter outputs: ensure filter_level_1_nb returns exactly 3 items and filter_level_2_nb returns exactly 1 item.
• Watermark Propagation: After a successful run, query config.curated_control_tbl. Verify that all notebook_levels have updated last_pipeline_runtime timestamps, with Level 1 timestamps slightly earlier than Level 2.

Data Quality Tests

• No Duplicates: After running the pipeline twice, count rows in curated.cab_trips_f. The count should equal the count of unique trip_ids in cab_landing, not double (no duplicates from Level 2 re-processing).
• Transient Cleanup: Verify that transient tables are overwritten (not appended) on each run. Count transient.cab_trips_f before and after a run where landing has 0 new data. The count should remain the same.
• Join Integrity: Query curated.cab_trips_f. Verify that every row has a valid trip_id that exists in curated.com_trips_header_f (referential integrity).
• Watermark Filtering: Manually update the watermark in config.curated_control_tbl to tomorrow's date. Run the pipeline. Verify that Level 2 notebooks find 0 rows and exit cleanly (the dependent watermark filter prevented re-processing).

Performance Tests

• Baseline Run: Execute the full pipeline once. Record the duration of each major activity. This is your baseline.
• Parallelism Impact: Run with sequential execution enabled (change Sequential Run to TRUE in ForEach1). Compare total time. Parallel should be measurably faster.
• Scale with Config: Add 2 more Level 1 notebooks to the config. Run the pipeline. Total time should not increase much if parallelism is working (only the slowest Level 1 notebook determines ForEach1 duration).

Common Orchestration Mistakes

Mistake 1: Running ForEach1 and ForEach2 in Parallel – If you enable parallelism at the top level, Level 2 might start before Level 1 finishes, reading stale transient data. Always use sequential containers for level separation.

Mistake 2: Forgetting the Dependent Watermark Read in Level 2 – Level 2 notebooks should read the watermark of their dependencies (Level 1), not their own watermark. This ensures they only process new data from their dependencies.

Mistake 3: No Early Exit Logic in Notebooks – If a notebook finds 0 rows, it should exit early instead of creating empty tables. This prevents cascading 0-row outputs down the pipeline.

Mistake 4: Not Updating Watermarks in SP – If the stored procedure doesn't update the watermark for a notebook, the next run will re-process the same data. Always verify SP completion and log watermark updates.

What's Next?

With the curated layer ready, data is now clean, deduplicated, and joined. In Part 7, we dive into the transformation notebooks themselves: how to write efficient PySpark code for the scale of data you're likely to encounter, how to optimize DataFrame operations, and how to structure code for testability and maintainability.

Rohit Ram

Senior Data Analyst · LinkedIn

Building production-grade data platforms on Microsoft Fabric. This series documents a real project — every SQL query, every design decision, every challenge encountered and solved.