1. DAG Design Principles (+3)
1. DAG Design Principles
# Use meaningful DAG and task IDs
dag_id='sales_daily_etl_pipeline' # Good
dag_id='dag1' # Bad
# Set appropriate concurrency limits
max_active_runs=1 # For data pipelines with dependencies
max_active_tasks_per_dag=16 # Limit resource usage
# Use tags for organization
tags=['production', 'etl', 'sales']
# Always set catchup=False unless backfill needed
catchup=False
# Use execution_timeout to prevent stuck tasks
execution_timeout=timedelta(hours=2)
2. Task Best Practices
# Keep tasks atomic and idempotent
def process_partition(partition_date: str):
"""Idempotent: can be safely re-run."""
# Delete existing data for this partition
delete_partition(partition_date)
# Process and insert new data
insert_data(partition_date)
# Use retries with exponential backoff
default_args = {
'retries': 3,
'retry_delay': timedelta(minutes=5),
'retry_exponential_backoff': True,
}
# Avoid heavy processing in sensors
# Bad: sensor does complex computation
# Good: sensor checks simple condition, processing in separate task
3. Configuration Management
# Use Variables for configuration, not hardcoded values
batch_size = Variable.get('batch_size', default_var=1000)
# Use Connections for credentials
conn = BaseHook.get_connection('my_database')
# Environment-specific configuration
env = Variable.get('environment')
config = Variable.get(f'config_{env}', deserialize_json=True)
4. Testing DAGs
# tests/test_dags.py
import pytest
from airflow.models import DagBag
def test_dag_loads():
"""Test that DAGs load without errors."""
dagbag = DagBag()
assert len(dagbag.import_errors) == 0
def test_dag_structure():
"""Test DAG has expected structure."""
dagbag = DagBag()
dag = dagbag.get_dag('my_pipeline')
assert dag is not None
assert len(dag.tasks) == 5
assert dag.schedule_interval == '@daily'
vamseeachanta