Steps That Merge Many Inputs

Concepts covered: Dependency Management, DAG Orchestration

In the curated layer, orders_enriched joins 3 inputs: curated orders (hourly, from Kafka), the product catalog (Debezium change capture out of Postgres, landing every 15 minutes) and FX rates (1 daily file from the European Central Bank, published around 16:00 CET). A step with 3 inputs faces 3 questions a single-input step never does: when it may run, which version of each input it reads, and what it does when 1 input has not arrived.

Cadence collapses to the slowest input

The join can run for hour 14 only when every input it needs for hour 14 exists. Orders land by 14:05 and catalog changes by 14:15, so the join could start at 14:15. The FX rate arrives once a day, so an hourly join at 03:00 UTC either waits 12 hours for today's rate or reads yesterday's. The orchestrator makes the wait explicit: in Airflow, an ExternalTaskSensor per upstream DAG or a Dataset schedule on all 3; in Dagster, 3 declared upstream assets. Either way, the step's effective cadence is the slowest of its inputs unless the design names which inputs may be stale and by how much.

Input	Cadence	Typically lands	Staleness the join accepts
Curated orders	hourly	:05 past the hour	None, the run waits
Product catalog (SCD2)	every 15 minutes	:15 past the hour	Up to 1 hour, prices change rarely
FX rates	daily	15:00 UTC	Previous day's rate until today's lands, rate date recorded

Three inputs at 3 cadences converge on 1 transform. The hourly edge gates the run, the 15-minute edge may be up to 1 hour stale, and the daily edge supplies yesterday's value until today's arrives.

curated orders (silver, Snowflake, hourly)
catalog SCD2 (silver, Snowflake, 15 min)
fx rates (silver, Snowflake, daily)
enrich join (transform, dbt model); freshness: hourly, by :20; backfill: rerun every interval that read a changed input
orders_enriched (storage, Snowflake)

curated orders -> enrich join
catalog SCD2 -> enrich join
fx rates -> enrich join
enrich join -> orders_enriched

Which version of each input

A join that reads the current catalog gives a different answer on a rerun than it gave the first time. A backfill of March 12 run on March 20 would price March 12 orders at March 20 prices and file them under categories renamed in between. A merge step needs an as-of rule per input, each tied to the run's data_interval_start: orders whose order_hour equals the interval, catalog rows whose valid_from and valid_to bracket the interval's start, and the FX rate for the interval's date. With all 3 rules the step is reproducible, and reproducibility is what makes its retries and backfills safe.

SELECT o.order_id, o.amount, c.category,
       o.amount * fx.rate_to_usd AS amount_usd,
       fx.rate_date
FROM curated_orders o
LEFT JOIN catalog_scd2 c
  ON c.sku = o.sku
 AND c.valid_from <= '2024-03-15 14:00:00'
 AND (c.valid_to > '2024-03-15 14:00:00' OR c.valid_to IS NULL)
LEFT JOIN fx_rates fx
  ON fx.currency = o.currency
 AND fx.rate_date <= '2024-03-15'
WHERE o.order_hour = '2024-03-15 14:00:00'
QUALIFY ROW_NUMBER() OVER (
  PARTITION BY o.order_id ORDER BY fx.rate_date DESC
) = 1;

What a missing input does to the output

An inner join on the catalog drops every order whose sku has not arrived, and the drop is silent: 40,000 orders in, 39,850 out, and the dashboard shows $3,600 less revenue than the payment processor settled. A left join keeps the 150 orders and leaves category null, which is visible and recoverable, because a trailing-window rebuild fills them in when the catalog catches up. The characteristic failure of a merge step is partial input dressed as complete input, so the step checks before it writes. A row-count assertion that orders_enriched for the hour equals curated orders for the hour guards against the inner-join drop. A null-rate check on category above 0.1% guards against the late catalog, whose 150 nulls are 0.375% of the hour.

The step also changes what an upstream rerun means. When the catalog CDC job is backfilled for a day, every join run that read that day must rerun, or the enriched table keeps pre-fix prices. Dagster's partitioned assets, with partition mappings, propagate that backfill to the dependent join partitions. Airflow Datasets record the dependency, but a Dataset update triggers 1 new join run at the current date, so the affected intervals still need an explicit backfill of the join DAG over the same dates. A merge step with 3 inputs is 3 places a backfill can start from and 1 place all of them must pass through.

The orders_enriched join for hour 03 UTC cannot find an FX rate for today, because the ECB publishes around 15:00 UTC. The finance dashboard needs hourly revenue in USD through the day. Which design serves it?

Put a sensor on the FX file so the hour 03 join waits until 15:00
Join on the most recent rate on or before the interval's date, record rate_date in the row, and restate the day's rows after the new rate lands
Inner join on today's rate and let unmatched orders fall out until the rate arrives
Move the whole join to a daily run at 16:00

About this lesson

This section is part of the Pipeline Anatomy: Intermediate lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.