Pipeline Anatomy: Intermediate

At a mid-size retailer the order pipeline begins as 1 chain. A Kafka topic orders.v1 feeds an hourly Spark job that writes Parquet to S3, and a Snowflake task loads fct_revenue_hourly for the finance dashboard. The first change request comes from marketing, which wants orders joined to ad spend, and the quickest path is a second job reading the same topic. Three months later the recommendations team builds a third. Each job drops test orders, casts amount from string to NUMERIC and deduplicates on order_id, and each does it a little differently.

The cost arrives as disagreement. For Tuesday hour 14 finance counts 40,000 orders, marketing counts 40,210 because its job never filtered the staff-discount test accounts, and the recommendations model trains on 39,600 because its deduplication keys on order_id plus updated_at and drops legitimate resubmissions. Three teams hold 3 numbers, and a meeting decides which one is real. New sources multiply the problem: once the Postgres product catalog, the Zendesk ticket API and the Google Ads spend export arrive as sources, 4 sources and 3 consumers wired point to point need 12 jobs, and the cleaning of each source is written in 3 of them.

One source and 1 consumer need no middle layer; the transform can shape data for its only reader, which is the beginner pipeline. The second consumer of the same source is the moment to split, because the cleaning either gets copied or gets shared, and copied cleaning drifts within weeks. The second source feeding the same consumer is the other trigger: conforming keys, such as the catalog's sku against the order line's product_code, is work every later consumer needs too. Either event turns N x M jobs into N + M, with 1 landing job per source and 1 serving job per consumer.

The retailer's curated layer holds both. Casting amount to NUMERIC, deduplicating on order_id and removing test accounts are ELT: dbt models over raw tables that keep every row, so a rule change reruns against history without touching the source. The model below does the cast, filter and deduplication for curated orders, and it runs in Snowflake in about 40 seconds over 1 hour of raw data.

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.

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.

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.

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.

Start at the 2 ends, because they set the requirements. Sources decide arrival behaviour: Kafka delivers within seconds and replays from any offset inside its 7 days of retention, while the Debezium connector follows the Postgres write-ahead log and fails its task when it loses its replication slot. Consumers decide freshness: the finance mart promises yesterday complete by 08:00, the feature store promises 15-minute freshness. Then find where edges converge. Here orders meet the catalog at curated_orders, the role orders_enriched played in section 2: its cadence is bounded by its slowest input, and every consumer inherits that bound.

For each node, 3 questions. Does it retry, and is its write safe to retry? The raw landing overwrites an hourly S3 prefix, so 3 retries leave 1 copy. The curated load deletes and inserts by order_hour in 1 transaction, so a retry after a timed-out commit removes its own earlier rows before loading. A load that appended instead would commit 40,000 rows, lose the acknowledgement, and commit 40,000 more on attempt 2, with every task green and finance seeing double revenue at 08:00. Does a failure block downstream or let stale data through? The gate blocks, so a bad hour never reaches the mart, at the price of a frozen dashboard. Who is paged? The monitorAlert on each node names the condition, and the owner column names the person.

Ownership changes hands where the layer changes. The checkout team owns the topic and its schema, the catalog team owns the Postgres tables, the data platform owns raw through the gate, and each consuming team owns its serving table. The boundary between layers is where the contract lives: the checkout team may add fields to orders.v1 but may not rename order_id, and the platform may change how curated_orders is built but not what a row means. A diagram without owners turns every cross-boundary incident into 20 minutes of finding out who to call.

The first design gives each consumer its own reader on orders.v1. The dashboard path is the hourly batch already built. The feature path is a 15-minute Spark Structured Streaming job that reads the topic, cleans, aggregates per customer and writes to Feast's offline store, with its own test-account filter, deduplication and cast of amount. It ships fast, because neither team waits on the other, and it recreates the drift section 0 opened with.

The second design keeps 1 raw landing and 1 curated orders table, and forks after it: 1 serving job aggregates curated orders into the finance mart, and 1 computes features from the same rows. The cost is that the curated layer must run at the faster consumer's cadence. An hourly curated table cannot feed a 15-minute feature refresh, so the curated load moves to a 15-minute micro-batch, and the finance mart, still hourly, reads 4 curated slices per run. A 15-minute MERGE over 10,000 orders completes in under 30 seconds on an X-Small Snowflake warehouse.

Point-in-time correctness decides it. A training set asks, for each order on March 12, what the customer's 7-day order count was at that moment. From curated orders with order_hour as event time, that is a window over 1 table. From a streaming path with its own deduplication, the answer depends on which late orders the stream happened to see, and the model trains on features the dashboard cannot reproduce. Forking at curated makes the model's orders match finance's by construction.

A consumer that needs latency the curated layer cannot provide reads the source directly. Fraud scoring at checkout needs a decision within 200 milliseconds of the order event, and a 15-minute micro-batch is 4,500 times too slow. It gets a Flink job on orders.v1 with minimal cleaning, plus a nightly job reconciling its decisions against curated orders. The early fork is deliberate and bounded, and its reconciliation cost is written into the design.