Practical patterns for metadata-driven orchestration in multi-entity data platforms
Multi-Tenant Airflow DAG Generation: Which DAGs to Generate and When
Practical patterns for metadata-driven orchestration in multi-entity data platforms
The Scalability Problem
Your architecture is elegant: SQL dependencies derived from transformation code, Airflow calling Python loaders that check metadata conditions, a generic data model serving multiple entities. But you’re hitting a wall.
For each entity, you generate all Airflow DAGs. Then the Python loader checks at runtime if the SQL statement should execute. This works, but it doesn’t scale:
- DAG explosion : 50 entities × 100 transformations = 5,000 DAGs
- Scheduler overhead : Airflow parses all DAGs, even when most won’t run
- False positives : Dashboards show failures for transformations that were never relevant
- Resource waste : Workers spin up just to check metadata and exit
The question isn’t “how do we make Airflow faster?” It’s “which DAGs should we generate in the first place?”
The Core Insight: Three Layers of DAG Generation
Based on MDDE patterns and real-world multi-tenant implementations, there are three complementary strategies:
┌─────────────────────────────────────────────────────────────────────┐
│ DAG GENERATION STRATEGIES │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ Strategy 1: CONTRACT-BASED DAG GENERATION │
│ ───────────────────────────────────────────── │
│ Generate DAGs only for what each entity actually uses │
│ → Reduces DAG count by 60-80% │
│ │
│ Strategy 2: HIERARCHICAL DAG STRUCTURE │
│ ───────────────────────────────────────── │
│ One coordination DAG triggers entity-specific sub-DAGs │
│ → Reduces scheduler parsing by 90% │
│ │
│ Strategy 3: DYNAMIC TASK MAPPING │
│ ───────────────────────────────── │
│ Single DAG with runtime-expanded tasks per entity │
│ → Best for homogeneous transformations │
│ │
└─────────────────────────────────────────────────────────────────────┘
Strategy 1: Contract-Based DAG Generation
The Problem with “Generate Everything”
Your current approach:
# Current: Generate all DAGs for all entities
for entity in all_entities:
for transformation in all_transformations:
generate_dag(entity, transformation) # 5,000 DAGs
The Python loader then checks at runtime:
def should_execute(entity, transformation, period):
contract = get_delivery_contract(entity)
if transformation not in contract.relevant_transformations:
return False # Skip - but DAG still exists!
The Solution: Generate Only Relevant DAGs
Use the Delivery Contract metadata to filter DAG generation at build time:
# Better: Generate only relevant DAGs per entity
for entity in all_entities:
contract = get_delivery_contract(entity)
sub_model = derive_sub_model(contract)
for transformation in sub_model.transformations:
generate_dag(entity, transformation) # 1,500 DAGs
MDDE Pattern: Entity Sub-Model Resolution
In MDDE, we model this as entity-specific sub-models derived from delivery contracts:
# delivery_contracts/bank_x.yaml
entity_id: bank_x
delivery_contract:
datasets:
- accounts
- customers
- transactions
excluded_datasets:
- cards # Bank X doesn't deliver cards
- investments # Not relevant for this entity
table_overrides:
accounts:
excluded_columns:
- legacy_account_type # Not delivered by this entity
The DAG generator reads this contract and generates only relevant DAGs:
def generate_entity_dags(entity_id: str) -> List[DAG]:
"""Generate DAGs based on entity's delivery contract."""
contract = load_delivery_contract(entity_id)
global_model = load_global_data_model()
# Derive sub-model from contract
sub_model = global_model.filter_by_contract(contract)
# Build dependency graph for sub-model only
dependency_graph = build_dependency_graph(sub_model)
# Generate DAGs for relevant transformations
dags = []
for transformation in sub_model.transformations:
dag = create_dag(
entity=entity_id,
transformation=transformation,
dependencies=dependency_graph.get_dependencies(transformation),
join_semantics=contract.get_join_semantics(transformation),
)
dags.append(dag)
return dags
Result
Strategy 2: Hierarchical DAG Structure
The Pattern: Coordination + Execution DAGs
Instead of flat DAGs per entity/transformation, use a two-level hierarchy:
┌─────────────────────────────────────────────────────────────────────┐
│ HIERARCHICAL DAG STRUCTURE │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ Level 1: COORDINATION DAGs (few, always parsed) │
│ ┌──────────────────────────────────────────────────────────────┐ │
│ │ daily_orchestrator │ │
│ │ ├── Check which entities have data ready │ │
│ │ ├── Trigger relevant entity DAGs │ │
│ │ └── Aggregate completion status │ │
│ └──────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ Level 2: EXECUTION DAGs (many, triggered on demand) │
│ ┌──────────────┐ ┌──────────────┐ ┌────────────────┐ │
│ │ bank_x_load │ │ bank_y_load │ │ fintech_z_load │ │
│ │ - accounts │ │ - accounts │ │ - accounts │ │
│ │ - customers │ │ - customers │ │ - cards │ │
│ │ - txns │ │ - txns │ │ - txns │ │
│ └──────────────┘ └──────────────┘ └────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘
Implementation
Level 1: Coordination DAG
# dags/daily_orchestrator.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
def check_entity_readiness(**context):
"""Check which entities have data ready for processing."""
execution_date = context['execution_date']
ready_entities = []
for entity in get_all_entities():
contract = get_delivery_contract(entity)
calendar = get_delivery_calendar(entity)
if calendar.expects_delivery(execution_date):
if check_source_completeness(entity, execution_date):
ready_entities.append(entity)
return ready_entities
with DAG(
dag_id="daily_orchestrator",
schedule="@daily",
tags=["orchestration", "coordinator"],
) as dag:
check_readiness = PythonOperator(
task_id="check_entity_readiness",
python_callable=check_entity_readiness,
)
# Dynamic task generation based on readiness
# (Using Airflow 2.3+ dynamic task mapping)
trigger_entities = TriggerDagRunOperator.partial(
task_id="trigger_entity_load",
wait_for_completion=True,
deferrable=True, # Free worker while waiting
).expand(
trigger_dag_id=check_readiness.output.map(lambda e: f"{e}_load"),
conf=check_readiness.output.map(lambda e: {"entity": e}),
)
Level 2: Entity Execution DAG
# dags/entity_dags/bank_x_load.py
def create_entity_dag(entity_id: str) -> DAG:
"""Create execution DAG for specific entity."""
contract = get_delivery_contract(entity_id)
sub_model = derive_sub_model(contract)
with DAG(
dag_id=f"{entity_id}_load",
schedule=None, # Triggered by coordinator
tags=["execution", entity_id],
) as dag:
# Generate tasks based on sub-model
tasks = {}
for transformation in sub_model.transformations:
tasks[transformation.id] = create_transformation_task(
entity_id=entity_id,
transformation=transformation,
join_semantics=contract.get_join_semantics(transformation),
)
# Wire dependencies from sub-model
for transformation in sub_model.transformations:
for dep in transformation.dependencies:
if dep.id in tasks:
tasks[dep.id] >> tasks[transformation.id]
return dag
# Generate DAGs for all entities
for entity in get_all_entities():
globals()[f"{entity}_load"] = create_entity_dag(entity)
Benefits
Strategy 3: Dynamic Task Mapping (Airflow 2.3+)
When to Use This
For homogeneous transformations where the same logic applies across entities, use a single DAG with dynamic task mapping:
# dags/generic_entity_loader.py
from airflow.decorators import dag, task
@dag(
dag_id="generic_entity_loader",
schedule="@daily",
tags=["multi-tenant"],
)
def generic_entity_loader():
@task
def get_ready_entities(execution_date) -> List[dict]:
"""Return entities ready for processing with their config."""
ready = []
for entity in get_all_entities():
contract = get_delivery_contract(entity)
calendar = get_delivery_calendar(entity)
if calendar.expects_delivery(execution_date):
completeness = check_source_completeness(entity, execution_date)
if completeness.inner_sources_ready:
ready.append({
"entity_id": entity,
"sub_model": derive_sub_model(contract),
"completeness": completeness,
})
return ready
@task
def load_entity(entity_config: dict):
"""Load data for a single entity."""
entity_id = entity_config["entity_id"]
sub_model = entity_config["sub_model"]
for transformation in topological_sort(sub_model.transformations):
execute_transformation(
entity_id=entity_id,
transformation=transformation,
)
return {"entity_id": entity_id, "status": "success"}
@task
def publish_completeness(results: List[dict]):
"""Update completeness dashboard."""
for result in results:
update_completeness_api(
entity=result["entity_id"],
status=result["status"],
)
# Wire it together with dynamic mapping
entities = get_ready_entities()
results = load_entity.expand(entity_config=entities)
publish_completeness(results)
generic_entity_loader()
Combining with Deferrable Operators
For Databricks SQL execution, use deferrable operators to avoid blocking workers:
from airflow.providers.databricks.operators.databricks_sql import DatabricksSqlOperator
@task
def load_entity_deferred(entity_config: dict):
"""Load entity using deferrable Databricks SQL."""
entity_id = entity_config["entity_id"]
sub_model = entity_config["sub_model"]
tasks = []
for transformation in sub_model.transformations:
sql = generate_transformation_sql(entity_id, transformation)
task = DatabricksSqlOperator(
task_id=f"{entity_id}_{transformation.id}",
databricks_conn_id="databricks_default",
sql=sql,
deferrable=True, # Free worker while Databricks processes
)
tasks.append(task)
return tasks
MDDE Integration: The Complete Pattern
1. Metadata Structure
# models/entities/bank_x/delivery_contract.yaml
entity:
id: bank_x
name: "Bank X"
delivery_contract:
# Which datasets this entity delivers
datasets:
- id: accounts
tables: [stg_accounts, stg_account_types]
frequency: daily
sla: "T+1 10:00 CET"
- id: customers
tables: [stg_customers, stg_addresses]
frequency: daily
sla: "T+1 10:00 CET"
- id: cards
status: excluded # Not delivered by this entity
reason: "Product not offered"
# Join semantics for enrichment
join_semantics:
accounts_enriched:
inner_joins: [stg_accounts, stg_customers] # Must have both
left_joins: [stg_cards, stg_preferences] # Optional enrichment
# Calendar overrides
calendar:
exceptions:
- date: "2026-01-01"
action: skip
reason: "Bank holiday"
2. Sub-Model Resolution
# src/mdde/orchestration/sub_model.py
from dataclasses import dataclass
from typing import List, Set
@dataclass
class SubModel:
"""Entity-specific subset of the global data model."""
entity_id: str
tables: Set[str]
transformations: List[Transformation]
dependencies: Dict[str, List[str]]
join_semantics: Dict[str, JoinSemantics]
@classmethod
def from_contract(cls, entity_id: str, contract: DeliveryContract, global_model: DataModel):
"""Derive sub-model from delivery contract."""
# Start with contracted datasets
relevant_tables = set()
for dataset in contract.datasets:
if dataset.status != "excluded":
relevant_tables.update(dataset.tables)
# Trace dependencies backwards from relevant tables
all_relevant = trace_dependencies(relevant_tables, global_model)
# Filter transformations to only relevant ones
transformations = [
t for t in global_model.transformations
if t.target in all_relevant
]
# Build dependency graph for sub-model
dependencies = {}
for t in transformations:
deps = [d for d in t.dependencies if d in all_relevant]
dependencies[t.id] = deps
return cls(
entity_id=entity_id,
tables=all_relevant,
transformations=transformations,
dependencies=dependencies,
join_semantics=contract.join_semantics,
)
3. DAG Generation Pipeline
# src/mdde/orchestration/dag_generator.py
def generate_dags_for_entity(entity_id: str) -> List[DAG]:
"""Generate Airflow DAGs for an entity based on its contract."""
# Load metadata
contract = load_delivery_contract(entity_id)
global_model = load_global_data_model()
calendar = load_delivery_calendar(entity_id)
# Derive sub-model
sub_model = SubModel.from_contract(entity_id, contract, global_model)
# Choose generation strategy based on sub-model characteristics
if sub_model.is_simple():
# Single DAG with all transformations
return [generate_single_dag(entity_id, sub_model)]
else:
# Multiple DAGs grouped by dataset
return generate_dataset_dags(entity_id, sub_model)
def generate_single_dag(entity_id: str, sub_model: SubModel) -> DAG:
"""Generate a single DAG for simple sub-models."""
with DAG(
dag_id=f"{entity_id}_load",
schedule=None, # Triggered by coordinator
tags=["execution", entity_id],
doc_md=generate_dag_documentation(entity_id, sub_model),
) as dag:
# Create tasks for each transformation
tasks = {}
for transformation in sub_model.transformations:
join_sem = sub_model.join_semantics.get(transformation.id)
tasks[transformation.id] = create_task(
transformation=transformation,
entity_id=entity_id,
join_semantics=join_sem,
deferrable=True,
)
# Wire dependencies
for t_id, deps in sub_model.dependencies.items():
for dep in deps:
if dep in tasks:
tasks[dep] >> tasks[t_id]
return dag
4. Runtime Completeness Checking
# src/mdde/orchestration/completeness.py
@dataclass
class CompletenessResult:
entity_id: str
period: date
inner_sources_ready: bool
left_sources_ready: bool
missing_inner: List[str]
missing_left: List[str]
can_publish_base: bool
can_publish_enriched: bool
def check_completeness(entity_id: str, period: date) -> CompletenessResult:
"""Check source completeness for an entity and period."""
contract = load_delivery_contract(entity_id)
sub_model = SubModel.from_contract(entity_id, contract, load_global_model())
# Check each source table
available = set()
for table in sub_model.tables:
if table_has_data(table, entity_id, period):
available.add(table)
# Separate by join semantics
for transformation in sub_model.transformations:
join_sem = sub_model.join_semantics.get(transformation.id)
if join_sem:
inner_ready = all(t in available for t in join_sem.inner_joins)
left_ready = all(t in available for t in join_sem.left_joins)
missing_inner = [t for t in join_sem.inner_joins if t not in available]
missing_left = [t for t in join_sem.left_joins if t not in available]
return CompletenessResult(
entity_id=entity_id,
period=period,
inner_sources_ready=inner_ready,
left_sources_ready=left_ready,
missing_inner=missing_inner,
missing_left=missing_left,
can_publish_base=inner_ready,
can_publish_enriched=inner_ready and left_ready,
)
Decision Matrix: Which Strategy When?
Implementation Checklist
Phase 1: Metadata Foundation (Week 1–2)
- [ ] Define delivery contract schema (YAML/JSON)
- [ ] Create contracts for top 5 entities
- [ ] Implement sub-model derivation logic
- [ ] Build completeness checking API
Phase 2: DAG Generation (Week 3–4)
- [ ] Implement contract-based DAG generator
- [ ] Create coordinator DAG
- [ ] Generate entity DAGs from contracts
- [ ] Add deferrable operators for Databricks
Phase 3: Runtime Optimization (Week 5–6)
- [ ] Implement dynamic task mapping for homogeneous loads
- [ ] Add completeness dashboard
- [ ] Configure per-entity resource pools
- [ ] Set up alerting based on contract SLAs
Phase 4: Scale & Refine (Ongoing)
- [ ] Extend contracts to all entities
- [ ] Tune Airflow scheduler settings
- [ ] Implement DAG versioning (Airflow 3.0)
- [ ] Add self-service contract editing
Airflow Configuration Tips
Based on your multi-tenant Databricks setup:
# airflow.cfg optimizations for multi-tenant
[scheduler]
# Reduce parsing overhead
min_file_process_interval = 60 # Parse DAGs less frequently
dag_dir_list_interval = 120 # Scan for new DAGs less often
parsing_processes = 4 # Parallel parsing
[core]
# Optimize for triggered DAGs
max_active_tasks_per_dag = 16 # Per-entity parallelism
max_active_runs_per_dag = 1 # One run per entity at a time
parallelism = 64 # Total concurrent tasks
[operators]
# Enable deferrable by default where available
default_deferrable = True
Closing Thoughts
The path from “generate all DAGs, check at runtime” to “generate only relevant DAGs, execute efficiently” involves three shifts:
- Metadata as the source of truth — Delivery contracts define what’s relevant, not code
- Build-time filtering — Generate only what’s needed, not everything possible
- Runtime intelligence — Completeness-aware triggering, not blind scheduling
Your Python loader checking metadata at runtime isn’t wrong — it’s just happening too late. Move that intelligence to DAG generation time, and your Airflow instance becomes a lean orchestrator instead of a overwhelmed babysitter.