airflow-dag-patterns

Name: airflow-dag-patterns
Author: wshobson

8.4kinstalls4.1kforks

How to install airflow-dag-patterns?

wshobson/agents/airflow-dag-patterns

$npx -y skills add wshobson/agents --skill airflow-dag-patterns

Installs into the current project.

Summary

Build production Apache Airflow DAGs with best practices for operators, sensors, testing, and deployment. Use when creating data pipelines, orchestrating workflows, or scheduling batch jobs.

Files · 1

View on GitHub

SKILL.md

1	# Apache Airflow DAG Patterns
2
3	Production-ready patterns for Apache Airflow including DAG design, operators, sensors, testing, and deployment strategies.
4
5	## When to Use This Skill
6
7	- Creating data pipeline orchestration with Airflow
8	- Designing DAG structures and dependencies
9	- Implementing custom operators and sensors
10	- Testing Airflow DAGs locally
11	- Setting up Airflow in production
12	- Debugging failed DAG runs
13
14	## Core Concepts
15
16	### 1. DAG Design Principles
17
18	\| Principle \| Description \|
19	\| --------------- \| ----------------------------------- \|
20	\| Idempotent \| Running twice produces same result \|
21	\| Atomic \| Tasks succeed or fail completely \|
22	\| Incremental \| Process only new/changed data \|
23	\| Observable \| Logs, metrics, alerts at every step \|
24
25	### 2. Task Dependencies
26
27	```python
28	# Linear
29	task1 >> task2 >> task3
30
31	# Fan-out
32	task1 >> [task2, task3, task4]
33
34	# Fan-in
35	[task1, task2, task3] >> task4
36
37	# Complex
38	task1 >> task2 >> task4
39	task1 >> task3 >> task4
40	```
41
42	## Quick Start
43
44	```python
45	# dags/example_dag.py
46	from datetime import datetime, timedelta
47	from airflow import DAG
48	from airflow.operators.python import PythonOperator
49	from airflow.operators.empty import EmptyOperator
50
51	default_args = {
52	'owner': 'data-team',
53	'depends_on_past': False,
54	'email_on_failure': True,
55	'email_on_retry': False,
56	'retries': 3,
57	'retry_delay': timedelta(minutes=5),
58	'retry_exponential_backoff': True,
59	'max_retry_delay': timedelta(hours=1),
60	}
61
62	with DAG(
63	dag_id='example_etl',
64	default_args=default_args,
65	description='Example ETL pipeline',
66	schedule='0 6 * * *', # Daily at 6 AM
67	start_date=datetime(2024, 1, 1),
68	catchup=False,
69	tags=['etl', 'example'],
70	max_active_runs=1,
71	) as dag:
72
73	start = EmptyOperator(task_id='start')
74
75	def extract_data(**context):
76	execution_date = context['ds']
77	# Extract logic here
78	return {'records': 1000}
79
80	extract = PythonOperator(
81	task_id='extract',
82	python_callable=extract_data,
83	)
84
85	end = EmptyOperator(task_id='end')
86
87	start >> extract >> end
88	```
89
90	## Detailed patterns and worked examples
91
92	Detailed pattern documentation lives in `references/details.md`. Read that file when the navigation tier above is insufficient.
93
94	## Best Practices
95
96	### Do's
97
98	- Use TaskFlow API - Cleaner code, automatic XCom
99	- Set timeouts - Prevent zombie tasks
100	- Use `mode='reschedule'` - For sensors, free up workers
101	- Test DAGs - Unit tests and integration tests
102	- Idempotent tasks - Safe to retry
103
104	### Don'ts
105
106	- Don't use `depends_on_past=True` - Creates bottlenecks
107	- Don't hardcode dates - Use `{{ ds }}` macros
108	- Don't use global state - Tasks should be stateless
109	- Don't skip catchup blindly - Understand implications
110	- Don't put heavy logic in DAG file - Import from modules

People who liked this also viewed

6 picks

Type

airflow-dag-patterns

8.4kinstalls4.1kforks

View on GitHub

How to install airflow-dag-patterns?

wshobson/agents/airflow-dag-patterns

$npx -y skills add wshobson/agents --skill airflow-dag-patterns

Installs into the current project.

Summary

Build production Apache Airflow DAGs with best practices for operators, sensors, testing, and deployment. Use when creating data pipelines, orchestrating workflows, or scheduling batch jobs.

Files · 1

View on GitHub

SKILL.md

1	# Apache Airflow DAG Patterns
2
3	Production-ready patterns for Apache Airflow including DAG design, operators, sensors, testing, and deployment strategies.
4
5	## When to Use This Skill
6
7	- Creating data pipeline orchestration with Airflow
8	- Designing DAG structures and dependencies
9	- Implementing custom operators and sensors
10	- Testing Airflow DAGs locally
11	- Setting up Airflow in production
12	- Debugging failed DAG runs
13
14	## Core Concepts
15
16	### 1. DAG Design Principles
17
18	\| Principle \| Description \|
19	\| --------------- \| ----------------------------------- \|
20	\| Idempotent \| Running twice produces same result \|
21	\| Atomic \| Tasks succeed or fail completely \|
22	\| Incremental \| Process only new/changed data \|
23	\| Observable \| Logs, metrics, alerts at every step \|
24
25	### 2. Task Dependencies
26
27	```python
28	# Linear
29	task1 >> task2 >> task3
30
31	# Fan-out
32	task1 >> [task2, task3, task4]
33
34	# Fan-in
35	[task1, task2, task3] >> task4
36
37	# Complex
38	task1 >> task2 >> task4
39	task1 >> task3 >> task4
40	```
41
42	## Quick Start
43
44	```python
45	# dags/example_dag.py
46	from datetime import datetime, timedelta
47	from airflow import DAG
48	from airflow.operators.python import PythonOperator
49	from airflow.operators.empty import EmptyOperator
50
51	default_args = {
52	'owner': 'data-team',
53	'depends_on_past': False,
54	'email_on_failure': True,
55	'email_on_retry': False,
56	'retries': 3,
57	'retry_delay': timedelta(minutes=5),
58	'retry_exponential_backoff': True,
59	'max_retry_delay': timedelta(hours=1),
60	}
61
62	with DAG(
63	dag_id='example_etl',
64	default_args=default_args,
65	description='Example ETL pipeline',
66	schedule='0 6 * * *', # Daily at 6 AM
67	start_date=datetime(2024, 1, 1),
68	catchup=False,
69	tags=['etl', 'example'],
70	max_active_runs=1,
71	) as dag:
72
73	start = EmptyOperator(task_id='start')
74
75	def extract_data(**context):
76	execution_date = context['ds']
77	# Extract logic here
78	return {'records': 1000}
79
80	extract = PythonOperator(
81	task_id='extract',
82	python_callable=extract_data,
83	)
84
85	end = EmptyOperator(task_id='end')
86
87	start >> extract >> end
88	```
89
90	## Detailed patterns and worked examples
91
92	Detailed pattern documentation lives in `references/details.md`. Read that file when the navigation tier above is insufficient.
93
94	## Best Practices
95
96	### Do's
97
98	- Use TaskFlow API - Cleaner code, automatic XCom
99	- Set timeouts - Prevent zombie tasks
100	- Use `mode='reschedule'` - For sensors, free up workers
101	- Test DAGs - Unit tests and integration tests
102	- Idempotent tasks - Safe to retry
103
104	### Don'ts
105
106	- Don't use `depends_on_past=True` - Creates bottlenecks
107	- Don't hardcode dates - Use `{{ ds }}` macros
108	- Don't use global state - Tasks should be stateless
109	- Don't skip catchup blindly - Understand implications
110	- Don't put heavy logic in DAG file - Import from modules

People who liked this also viewed

6 picks

Type