企业官网建设流程全解析

Apache Airflow DAG 模式实现手册

此文件包含技能引用的详细模式、清单和代码示例。

核心概念

1. DAG 设计原则

2. 任务依赖

# Lineartask1>>task2>>task3# Fan-outtask1>>[task2,task3,task4]# Fan-in[task1,task2,task3]>>task4# Complextask1>>task2>>task4 task1>>task3>>task4

快速开始

# dags/example_dag.pyfromdatetimeimportdatetime,timedeltafromairflowimportDAGfromairflow.operators.pythonimportPythonOperatorfromairflow.operators.emptyimportEmptyOperator default_args={'owner':'data-team','depends_on_past':False,'email_on_failure':True,'email_on_retry':False,'retries':3,'retry_delay':timedelta(minutes=5),'retry_exponential_backoff':True,'max_retry_delay':timedelta(hours=1),}withDAG(dag_id='example_etl',default_args=default_args,description='Example ETL pipeline',schedule='0 6 * * *',# Daily at 6 AMstart_date=datetime(2024,1,1),catchup=False,tags=['etl','example'],max_active_runs=1,)asdag:start=EmptyOperator(task_id='start')defextract_data(**context):execution_date=context['ds']# Extract logic herereturn{'records':1000}extract=PythonOperator(task_id='extract',python_callable=extract_data,)end=EmptyOperator(task_id='end')start>>extract>>end

模式

模式 1：TaskFlow API（Airflow 2.0）

# dags/taskflow_example.pyfromdatetimeimportdatetimefromairflow.decoratorsimportdag,taskfromairflow.modelsimportVariable@dag(dag_id='taskflow_etl',schedule='@daily',start_date=datetime(2024,1,1),catchup=False,tags=['etl','taskflow'],)deftaskflow_etl():"""ETL pipeline using TaskFlow API"""@task()defextract(source:str)->dict:"""Extract data from source"""importpandasaspd df=pd.read_csv(f's3://bucket/{source}/{{ ds }}.csv')return{'data':df.to_dict(),'rows':len(df)}@task()deftransform(extracted:dict)->dict:"""Transform extracted data"""importpandasaspd df=pd.DataFrame(extracted['data'])df['processed_at']=datetime.now()df=df.dropna()return{'data':df.to_dict(),'rows':len(df)}@task()defload(transformed:dict,target:str):"""Load data to target"""importpandasaspd df=pd.DataFrame(transformed['data'])df.to_parquet(f's3://bucket/{target}/{{ ds }}.parquet')returntransformed['rows']@task()defnotify(rows_loaded:int):"""Send notification"""print(f'Loaded{rows_loaded}rows')# Define dependencies with XCom passingextracted=extract(source='raw_data')transformed=transform(extracted)loaded=load(transformed,target='processed_data')notify(loaded)# Instantiate the DAGtaskflow_etl()

模式 2：动态 DAG 生成

# dags/dynamic_dag_factory.pyfromdatetimeimportdatetime,timedeltafromairflowimportDAGfromairflow.operators.pythonimportPythonOperatorfromairflow.modelsimportVariableimportjson# Configuration for multiple similar pipelinesPIPELINE_CONFIGS=[{'name':'customers','schedule':'@daily','source':'s3://raw/customers'},{'name':'orders','schedule':'@hourly','source':'s3://raw/orders'},{'name':'products','schedule':'@weekly','source':'s3://raw/products'},]defcreate_dag(config:dict)->DAG:"""Factory function to create DAGs from config"""dag_id=f"etl_{config['name']}"default_args={'owner':'data-team','retries':3,'retry_delay':timedelta(minutes=5),}dag=DAG(dag_id=dag_id,default_args=default_args,schedule=config['schedule'],start_date=datetime(2024,1,1),catchup=False,tags=['etl','dynamic',config['name']],)withdag:defextract_fn(source,**context):print(f"Extracting from{source}for{context['ds']}")deftransform_fn(**context):print(f"Transforming data for{context['ds']}")defload_fn(table_name,**context):print(f"Loading to{table_name}for{context['ds']}")extract=PythonOperator(task_id='extract',python_callable=extract_fn,op_kwargs={'source':config['source']},)transform=PythonOperator(task_id='transform',python_callable=transform_fn,)load=PythonOperator(task_id='load',python_callable=load_fn,op_kwargs={'table_name':config['name']},)extract>>transform>>loadreturndag# Generate DAGsforconfiginPIPELINE_CONFIGS:globals()[f"dag_{config['name']}"]=create_dag(config)

模式 3：分支与条件逻辑

# dags/branching_example.pyfromairflow.decoratorsimportdag,taskfromairflow.operators.pythonimportBranchPythonOperatorfromairflow.operators.emptyimportEmptyOperatorfromairflow.utils.trigger_ruleimportTriggerRule@dag(dag_id='branching_pipeline',schedule='@daily',start_date=datetime(2024,1,1),catchup=False,)defbranching_pipeline():@task()defcheck_data_quality()->dict:"""Check data quality and return metrics"""quality_score=0.95# Simulatedreturn{'score':quality_score,'rows':10000}defchoose_branch(**context)->str:"""Determine which branch to execute"""ti=context['ti']metrics=ti.xcom_pull(task_ids='check_data_quality')ifmetrics['score']>=0.9:return'high_quality_path'elifmetrics['score']>=0.7:return'medium_quality_path'else:return'low_quality_path'quality_check=check_data_quality()branch=BranchPythonOperator(task_id='branch',python_callable=choose_branch,)high_quality=EmptyOperator(task_id='high_quality_path')medium_quality=EmptyOperator(task_id='medium_quality_path')low_quality=EmptyOperator(task_id='low_quality_path')# Join point - runs after any branch completesjoin=EmptyOperator(task_id='join',trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS,)quality_check>>branch>>[high_quality,medium_quality,low_quality]>>join branching_pipeline()

模式 4：传感器与外部依赖

# dags/sensor_patterns.pyfromdatetimeimportdatetime,timedeltafromairflowimportDAGfromairflow.sensors.filesystemimportFileSensorfromairflow.providers.amazon.aws.sensors.s3importS3KeySensorfromairflow.sensors.external_taskimportExternalTaskSensorfromairflow.operators.pythonimportPythonOperatorwithDAG(dag_id='sensor_example',schedule='@daily',start_date=datetime(2024,1,1),catchup=False,)asdag:# Wait for file on S3wait_for_file=S3KeySensor(task_id='wait_for_s3_file',bucket_name='data-lake',bucket_key='raw/{{ ds }}/data.parquet',aws_conn_id='aws_default',timeout=60*60*2,# 2 hourspoke_interval=60*5,# Check every 5 minutesmode='reschedule',# Free up worker slot while waiting)# Wait for another DAG to completewait_for_upstream=ExternalTaskSensor(task_id='wait_for_upstream_dag',external_dag_id='upstream_etl',external_task_id='final_task',execution_date_fn=lambdadt:dt,# Same execution datetimeout=60*60*3,mode='reschedule',)# Custom sensor using @task.sensor decorator@task.sensor(poke_interval=60,timeout=3600,mode='reschedule')defwait_for_api()->PokeReturnValue:"""Custom sensor for API availability"""importrequests response=requests.get('https://api.example.com/health')is_done=response.status_code==200returnPokeReturnValue(is_done=is_done,xcom_value=response.json())api_ready=wait_for_api()defprocess_data(**context):api_result=context['ti'].xcom_pull(task_ids='wait_for_api')print(f"API returned:{api_result}")process=PythonOperator(task_id='process',python_callable=process_data,)[wait_for_file,wait_for_upstream,api_ready]>>process

模式5：错误处理与警报

# dags/error_handling.pyfromdatetimeimportdatetime,timedeltafromairflowimportDAGfromairflow.operators.pythonimportPythonOperatorfromairflow.utils.trigger_ruleimportTriggerRulefromairflow.modelsimportVariabledeftask_failure_callback(context):"""Callback on task failure"""task_instance=context['task_instance']exception=context.get('exception')# Send to Slack/PagerDuty/etcmessage=f""" Task Failed! DAG:{task_instance.dag_id}Task:{task_instance.task_id}Execution Date:{context['ds']}Error:{exception}Log URL:{task_instance.log_url}"""# send_slack_alert(message)print(message)defdag_failure_callback(context):"""Callback on DAG failure"""# Aggregate failures, send summarypasswithDAG(dag_id='error_handling_example',schedule='@daily',start_date=datetime(2024,1,1),catchup=False,on_failure_callback=dag_failure_callback,default_args={'on_failure_callback':task_failure_callback,'retries':3,'retry_delay':timedelta(minutes=5),},)asdag:defmight_fail(**context):importrandomifrandom.random()<0.3:raiseValueError("Random failure!")return"Success"risky_task=PythonOperator(task_id='risky_task',python_callable=might_fail,)defcleanup(**context):"""Cleanup runs regardless of upstream failures"""print("Cleaning up...")cleanup_task=PythonOperator(task_id='cleanup',python_callable=cleanup,trigger_rule=TriggerRule.ALL_DONE,# Run even if upstream fails)defnotify_success(**context):"""Only runs if all upstream succeeded"""print("All tasks succeeded!")success_notification=PythonOperator(task_id='notify_success',python_callable=notify_success,trigger_rule=TriggerRule.ALL_SUCCESS,)risky_task>>[cleanup_task,success_notification]

模式 6：测试有向无环图

# tests/test_dags.pyimportpytestfromdatetimeimportdatetimefromairflow.modelsimportDagBag@pytest.fixturedefdagbag():returnDagBag(dag_folder='dags/',include_examples=False)deftest_dag_loaded(dagbag):"""Test that all DAGs load without errors"""assertlen(dagbag.import_errors)==0,f"DAG import errors:{dagbag.import_errors}"deftest_dag_structure(dagbag):"""Test specific DAG structure"""dag=dagbag.get_dag('example_etl')assertdagisnotNoneassertlen(dag.tasks)==3assertdag.schedule_interval=='0 6 * * *'deftest_task_dependencies(dagbag):"""Test task dependencies are correct"""dag=dagbag.get_dag('example_etl')extract_task=dag.get_task('extract')assert'start'in[t.task_idfortinextract_task.upstream_list]assert'end'in[t.task_idfortinextract_task.downstream_list]deftest_dag_integrity(dagbag):"""Test DAG has no cycles and is valid"""fordag_id,dagindagbag.dags.items():assertdag.test_cycle()isNone,f"Cycle detected in{dag_id}"# Test individual task logicdeftest_extract_function():"""Unit test for extract function"""fromdags.example_dagimportextract_data result=extract_data(ds='2024-01-01')assert'records'inresultassertisinstance(result['records'],int)

项目结构

airflow/ ├── dags/ │ ├── __init__.py │ ├── common/ │ │ ├── __init__.py │ │ ├── operators.py # Custom operators │ │ ├── sensors.py # Custom sensors │ │ └── callbacks.py # Alert callbacks │ ├── etl/ │ │ ├── customers.py │ │ └── orders.py │ └── ml/ │ └── training.py ├── plugins/ │ └── custom_plugin.py ├── tests/ │ ├── __init__.py │ ├── test_dags.py │ └── test_operators.py ├── docker-compose.yml └── requirements.txt

最佳实践

应做的事情

• 使用 TaskFlow API- 更清晰的代码，自动 XCom
• 设置超时- 防止僵尸任务
• 使用mode='reschedule'- 对于传感器，释放工作线程
• 测试 DAGs- 单元测试和集成测试
• 幂等任务- 安全可重试

禁忌

• 不要使用depends_on_past=True- 会造成瓶颈
• 不要硬编码日期- 使用{{ ds }}宏
• 不要使用全局状态- 任务应该是无状态的
• 不要盲目追赶- 了解其影响
• 不要在 DAG 文件中放置大量逻辑- 从模块中导入

资源

• Airflow Documentation
• Astronomer Guides
• TaskFlow API