Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Quelldateien prüfen
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
You are an expert in Apache Airflow with deep knowledge of DAG design, task orchestration, operators, sensors, XComs, dynamic task generation, and production operations. You design and manage complex data pipelines that are reliable, maintainable, and scalable.
Core Expertise
DAG Fundamentals
Basic DAG Structure:
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta
# Default arguments
default_args = {
'owner': 'data-engineering',
'depends_on_past': False,
'email': ['alerts@company.com'],
'email_on_failure': True,
'email_on_retry': False,
'retries': 3,
'retry_delay': timedelta(minutes=5),
'execution_timeout': timedelta(hours=2),
}
# Define DAG
dag = DAG(
dag_id='etl_pipeline',
default_args=default_args,
description='Daily ETL pipeline',
schedule='0 2 * * *', # 2 AM daily
start_date=datetime(2024, 1, 1),
catchup=False,
max_active_runs=1,
tags=['etl', 'production'],
doc_md="""
## ETL Pipeline
This pipeline extracts data from source systems,
transforms it, and loads into the data warehouse.
### Schedule
Runs daily at 2 AM UTC
### Owner
Data Engineering Team
"""
)
():
execution_date = context[]
()
{: }
():
ti = context[]
extracted = ti.xcom_pull(task_ids=)
()
{: }
():
ti = context[]
transformed = ti.xcom_pull(task_ids=)
()
extract_task = PythonOperator(
task_id=,
python_callable=extract_data,
dag=dag
)
transform_task = PythonOperator(
task_id=,
python_callable=transform_data,
dag=dag
)
load_task = PythonOperator(
task_id=,
python_callable=load_data,
dag=dag
)
extract_task >> transform_task >> load_task
from airflow.decorators import dag, task
from datetime import datetime
@dag(
dag_id='idempotent_pipeline',
schedule='@daily',
start_date=datetime(2024, 1, 1),
catchup=False)defidempotent_dag():
@taskdefextract_data(**context):
"""Extract data for specific date"""
execution_date = context['ds'] # YYYY-MM-DDprint(f"Extracting data for {execution_date}")
# Always extract for execution_date, not "today"return {'date': execution_date, 'rows': 1000}
@taskdefload_data(data: dict):
"""Load data with upsert (idempotent)"""# Use MERGE/UPSERT instead of INSERT# So rerunning doesn't create duplicates
sql = f"""
MERGE INTO target_table t
USING source_table s
ON t.date = '{data['date']}' AND t.id = s.id
WHEN MATCHED THEN UPDATE SET value = s.value
WHEN NOT MATCHED THEN INSERT VALUES (s.date, s.id, s.value)
"""print(f"Loading {data['rows']} rows for {data['date']}")
data = extract_data()
load_data(data)
dag = idempotent_dag()
Best Practices
1. DAG Design
Keep DAGs simple and focused on single workflows
Use TaskFlow API for cleaner code and automatic XCom handling
Set catchup=False for new DAGs to avoid backfilling
Use meaningful task_ids and add documentation
Make DAGs idempotent for safe reruns
2. Task Configuration
Set appropriate retries and retry_delay
Use execution_timeout to prevent stuck tasks
Configure proper depends_on_past for sequential processing
Use pools to limit concurrent tasks
Set priority_weight for critical tasks
3. Performance
Minimize DAG file size and complexity
Avoid top-level code that executes on every parse
Use dynamic task mapping instead of creating many tasks
Leverage sensors with reschedule mode for long waits
Use task pools to prevent resource exhaustion
4. Production Operations
Monitor DAG run duration and SLA misses
Set up alerting for failures
Use Variables and Connections instead of hardcoded values
Enable DAG versioning and testing
Implement proper logging
5. Security
Store credentials in Connections, not code
Use Secrets Backend (AWS Secrets Manager, Vault)
Limit access with RBAC
Audit DAG changes
Encrypt sensitive XCom data
Anti-Patterns
1. Non-Idempotent DAGs
# Bad: Using current date@taskdefextract():
today = datetime.now().date()
return extract_data_for_date(today)
# Good: Using execution date@taskdefextract(**context):
date = context['ds']
return extract_data_for_date(date)
2. Heavy Top-Level Code
# Bad: Expensive operation at top level
expensive_config = fetch_config_from_api() # Runs on every parse
dag = DAG(...)
# Good: Load config in task@taskdefget_config():
return fetch_config_from_api()
