When should your pipelines run? Not on a schedule — when there’s new data to process.
State-Aware Orchestration in Metadata-Driven Pipelines
When should your pipelines run? Not on a schedule — when there’s new data to process.
The Problem with Scheduled Pipelines
Hugo Lu’s article on State-Aware Orchestration identifies a fundamental issue: scheduled pipelines waste resources when:
- Data doesn’t land on time
- No new data has arrived since last run
- Models are forgotten but keep running
- Full refreshes happen when incremental would suffice
His framing of “5 assumptions that break at scale” is useful:
State-aware orchestration addresses this by tracking what was processed and only running when there’s work to do.
What Is State?
State in this context is metadata about what has been processed:
┌──────────────────────────────────────────────────────────────┐
│ STATE EXAMPLES │
├──────────────────────────────────────────────────────────────┤
│ │
│ HIGH WATERMARK │
│ {"customer_table": {"last_id": 45678, "last_ts": "..."}} │
│ │
│ CHECKPOINT │
│ {"pipeline": "customer_etl", "position": "offset:12345"} │
│ │
│ MANIFEST │
│ {"model": "customer", "hash": "abc123", "modified": false} │
│ │
│ PARTITION STATE │
│ {"table": "events", "partitions_loaded": ["2024-01", ...]} │
│ │
└──────────────────────────────────────────────────────────────┘
The insight: state has always existed. Kafka offsets, Spark checkpoints, Iceberg snapshots, Delta transaction logs — all are forms of state. dbt’s “state:modified” is just making this explicit for transformations.
State-Aware vs Schedule-Aware
Traditional scheduling:
# Run at 2am regardless of whether data exists
schedule:
cron: "0 2 * * *"
tasks:
- dbt run --select customer_model
State-aware:
# Run when source data has changed
model:
name: customer_model
sla: hourly
depends_on:
- source: crm.customers
state: watermark
The orchestrator:
- Checks if
crm.customershas new data (watermark > last processed) - If yes → run the model
- If no → skip (no wasted compute)
How This Relates to MDDE
MDDE already tracks much of the metadata needed for state-aware orchestration:
1. Entity Lineage = Dependency Graph
entity:
id: customer_360
sources:
- entity: crm_customer
- entity: erp_customer
- entity: web_analytics_customer
This defines what depends on what.
The orchestrator can use this to determine run order.
2. Delivery Properties = SLA Definition
entity:
id: customer_360
delivery_properties:
refresh_frequency: hourly
sla: "by 08:00 daily"
criticality: business_critical
This is the when it needs to be ready , not when to run it.
3. Source Freshness = State Tracking
source:
id: crm_customer
freshness:
loaded_at_field: _load_timestamp
warn_after: 12 hours
error_after: 24 hours
This is input state — has new data arrived?
4. Historization Metadata = Output State
The bi-temporal SCD2 pattern already tracks state:
-- What was last processed?
SELECT MAX(load_dts) FROM customer_scd2;
-- What's the current watermark?
SELECT MAX(valid_from) FROM customer_scd2 WHERE is_current = TRUE;
Adding State-Aware Orchestration to MDDE
New: Pipeline State Entity
entity:
id: pipeline_state
stereotype: operational_metadata
description: Tracks processing state for each entity
attributes:
- name: entity_id
type: string
description: The entity being tracked
- name: last_source_watermark
type: timestamp
description: Highest watermark seen in source
- name: last_processed_watermark
type: timestamp
description: Highest watermark actually processed
- name: last_run_timestamp
type: timestamp
description: When the pipeline last ran
- name: last_run_status
type: string
description: SUCCESS, FAILED, SKIPPED
- name: rows_processed
type: bigint
description: Number of rows in last run
- name: model_hash
type: string
description: Hash of model definition (detect schema changes)
New: SLA-Based Scheduling
# Instead of cron schedules, define SLAs
entity:
id: customer_360
orchestration:
sla:
- name: hourly_refresh
deadline: "+1 hour" # Ready within 1 hour of source data
triggers:
- type: source_change
sources: [crm_customer, erp_customer]
- type: schedule
cron: "0 * * * *" # Check hourly, but skip if no new data
skip_conditions:
- no_source_changes
- within_sla # Already met SLA, don't re-run
New: Watermark Configuration
source:
id: crm_customer
watermark:
type: timestamp
column: updated_at
# Or for systems without timestamps
# type: hash
# columns: [customer_id, customer_name, email]
The State Check Pattern
Before running any pipeline:
-- 1. Get current source watermark
WITH source_state AS (
SELECT MAX(updated_at) AS current_watermark
FROM source.crm_customers
),
-- 2. Get last processed watermark
processed_state AS (
SELECT last_processed_watermark
FROM pipeline_state
WHERE entity_id = 'customer_360'
),
-- 3. Determine if run is needed
run_decision AS (
SELECT
CASE
WHEN source_state.current_watermark > processed_state.last_processed_watermark
THEN TRUE
ELSE FALSE
END AS should_run,
source_state.current_watermark AS new_watermark
FROM source_state, processed_state
)
SELECT * FROM run_decision;
If should_run = FALSE, skip the pipeline entirely.
Integration with Bi-Temporal SCD2
The SCD2 pattern naturally supports state tracking:
SOURCE DATA STATE TARGET SCD2
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────────┐
│ id: 1 │ │ watermark: │ │ id: 1 │
│ name: "Acme" │ │ 2024-03-01 │ │ name: "Acme" │
│ updated_at: │ │ │ │ valid_from: 2024-01 │
│ 2024-03-01 │ │ last_run: │ │ valid_to: 9999-12 │
│ │ │ 2024-03-01 │ │ load_dts: 2024-03 │
└─────────────────┘ │ 06:00:00 │ └─────────────────────┘
└─────────────────┘
No new data scenario:
Source watermark: 2024-03-01
Last processed: 2024-03-01
Decision: SKIP (no new data)
New data scenario:
Source watermark: 2024-03-02
Last processed: 2024-03-01
Decision: RUN (process 2024-03-02 data)
State-Aware vs End-to-End Orchestration
Hugo makes an important point: end-to-end orchestration often eliminates the need for state tracking.
WITH END-TO-END ORCHESTRATION:
Source System ──► Ingestion ──► Transformation ──► Delivery
│ │ │ │
└──────────────┴───────────────┴───────────────┘
Single Pipeline
If ingestion finds no data → entire pipeline stops
No need for state checks downstream
This is the pattern MDDE already supports with unified pipeline definitions:
pipeline:
id: customer_etl
stages:
- id: ingest
source: crm_api
target: raw.crm_customers
on_no_data: stop # Pipeline ends here if no data
- id: historize
source: raw.crm_customers
target: source_hist.crm_customers
depends_on: [ingest]
- id: integrate
source: source_hist.crm_customers
target: integration.customer
depends_on: [historize]
When to add state-aware orchestration on top:
The IDS Pattern and State
The IDS (Integrated Data Sets) pattern from our previous article has natural state boundaries:
DELIVERY ENTITY A (IDS) DELIVERY ENTITY B (IDS)
│ │
│ State: watermark_a │ State: watermark_b
▼ ▼
┌─────────────────────────────────────────────────────────────┐
│ UNIFIED TABLE │
│ │
│ PARTITION: entity_a, SOURCE_IDS │
│ └─ Last processed: 2024-03-01 06:00 │
│ │
│ PARTITION: entity_b, SOURCE_IDS │
│ └─ Last processed: 2024-03-01 08:00 │
│ │
│ PARTITION: entity_a, ENRICHMENT │
│ └─ Last processed: 2024-03-01 07:00 │
│ │
└─────────────────────────────────────────────────────────────┘
Each partition can have independent state , enabling:
- Entity A updates without affecting Entity B
- Enrichment runs only when base IDS changes
- Fine-grained watermark tracking per source
Configuration: State-Aware Pipeline
# orchestration/customer_pipeline.yaml
pipeline:
id: customer_pipeline
state_tracking:
enabled: true
storage: pipeline_state_table # Or redis, file, etc.
sources:
- id: crm_customer
watermark:
column: updated_at
type: timestamp
- id: erp_customer
watermark:
column: modification_timestamp
type: timestamp
targets:
- id: customer_360
partition_by: [delivery_entity_id, source_type]
orchestration:
mode: state_aware # vs 'scheduled' or 'triggered'
check_interval: "*/15 * * * *" # Check every 15 min
run_conditions:
- any_source_has_new_data
- model_definition_changed
- manual_trigger
skip_conditions:
- all_sources_unchanged
- target_fresher_than_sla
sla:
deadline: "08:00 daily"
alert_on_miss: true
State Storage Options
MDDE approach: store state as a regular entity with the same patterns as business data:
entity:
id: pipeline_state
stereotype: operational_metadata
storage:
type: delta_table
location: metadata.pipeline_state
This means state is:
- Versioned (SCD2 if needed)
- Queryable
- Part of the same governance framework
- Backed up with other metadata
Impact-Driven Refresh Integration
State-aware orchestration connects to MDDE’s impact-driven refresh pattern:
# From ADR-275: Impact Analysis
impact_refresh:
entity_id: customer_360
# Only refresh if impacted sources have new data
impact_sources:
- crm_customer
- erp_customer
# Check watermarks before running
state_check:
- source: crm_customer
watermark: updated_at
- source: erp_customer
watermark: modification_timestamp
# Skip refresh if no impact
skip_if_no_impact: true
Key Takeaways
- State-aware orchestration tracks what was processed to avoid redundant runs
- MDDE already has the metadata needed (lineage, delivery properties, freshness)
- Bi-temporal SCD2 naturally supports watermark tracking via
load_dts - End-to-end orchestration often eliminates state needs for simple pipelines
- State as an entity makes it part of the governed metadata framework
- IDS + unified tables enable per-partition state tracking
- SLA-based scheduling is more declarative than cron expressions
The goal: pipelines that run when needed, not when scheduled.
This article is part of the Model-Driven Data Engineering series. Inspired by Hugo Lu’s State-Aware Orchestration article.