1. 前言
近期计划做一个任务调度系统，于是，重拾airflow，借机深入学习下。
主要调研和测试具体使用方法、能否满足我们的项目需求，以及可能存在哪些坑。
不了解airflow的朋友，可以参考我的上篇文章：
Python - Airflow任务调度系统初识
简单回顾一下两组关键名词：
- Dag -> DagRun(Dag Instance)
- Operator -> Task -> Task Instance
Core Concepts — Airflow Documentation (apache.org)
2. 使用方法
2.1. 编写Dag文件，并测试
image.png
- 梳理实际用户需求，是否存在选择分支，是否存在人工审核步骤，是否存在任务重跑，等
- 通过不同的Operator，实现一系列Task，以及任务的入参/出参、任务间通信、任务间依赖等（附件采用PythonOperator，实现了一个ETL的过程，采用xcom通信）
- 在Test环境，测试Dag文件，是否符合预期
比较常用的两个Operator：
- PythonOperator：可以通过args、kwargs设置Task运行时参数，通过xcom完成Task间通信
- BranchOperator：可以在运行时，决定走哪个任务分支，比如时间、服务器资源、上个任务的输出、当前Dag的状态等
坑
不采用json、yaml描述dag，便于airflow实现功能更加丰富的任务流，以及更自由的配置。
但是，如果让非软件开发人员编写Dag文件，无疑压力巨大。因此，需要人工审核。
如果是个2C软件，则可能需要jinja2来实时render一个dag.py文件。
Best Practices — Airflow Documentation (apache.org)
2.2. 上传Dag文件到scheduler和worker节点
image.png
Dag文件目录，需要在scheduler和worker机器上各存储一份，这个是由Airflow系统架构决定。
采用的方法，可以：
- nas盘，挂载到同一个集群的不同机器上（最好在一个局域网内）
- ansible手动push到服务器节点
- 在服务器节点，启动脚本，实时或定时同步oss/gitlab文件到本地
坑
如果Dag文件很多的话，文件的管理，可能需要namespace、mainfest等。
另外，是否需要先同步到worker节点，再同步到scheduler节点。
2.3. trigger dag
- 自动trigger，例如cron、upstream-task
- 手动trigger，通过命令行、airflow browser、airflow restful api
2.4. backfill
如果dag的下面两个参数为True，则dag一启动，则会查看过去的dagrun，如果存在未执行的dagrun，则回填。
所以，一般情况下，这两个参数需要手动设置为False。
depends_on_past=False
catchup=False
3. 日常维护
- metadata数据过多，会导致系统响应缓慢，通过airflow browser就可以明显感觉到，需要手动清理db
- 任务告警与相关责任人的管理
- 任务流上线流程，需要严格测试，并增加权限控制
- 任务流在生产环境运行时，需要引入pool的概念，来隔离不同任务流之间的资源竞争
- Scheduler只能跑一份，所以存在单点故障风险。需要借助第三方组件airflow-scheduler-failover-controller实现Scheduler的高可用
4. 参考
微信公众号【三帮大数据】的系列文章大数据调度平台Airflow
5. 附，ETL脚本
import json
import pendulum
from airflow import DAG
from airflow.operators.python import PythonOperator
with DAG(
"cp_pipeline_1",
default_args={"retries": 2},
description="computing pipeline",
schedule=None,
start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
depends_on_past=False,
catchup=False,
tags=["computing-platform"],
) as dag:
def extract(*args, **kwargs):
print('extract kwargs', kwargs)
ti = kwargs["ti"]
data_string = '{"1001": 301.27, "1002": 433.21, "1003": 502.22}'
ti.xcom_push("order_data", data_string)
return {'extract', 'over'} # return值也会进入到xcom，downstream的task也可以通过xcom获取
def transform(*args, **kwargs):
print('transform kwargs', kwargs)
ti = kwargs["ti"]
extract_data_string = ti.xcom_pull(task_ids="extract", key="order_data")
order_data = json.loads(extract_data_string)
total_order_value = 0
for value in order_data.values():
total_order_value += value
total_value = {"total_order_value": total_order_value}
total_value_json_string = json.dumps(total_value)
ti.xcom_push("total_order_value", total_value_json_string)
def load(*args, **kwargs):
print('load kwargs', kwargs)
ti = kwargs["ti"]
total_value_string = ti.xcom_pull(task_ids="transform", key="total_order_value")
total_order_value = json.loads(total_value_string)
print('load result', total_order_value)
extract_task = PythonOperator(
task_id="extract",
python_callable=extract,
op_args=[],
op_kwargs=dict(a1=1,b1='bb',c1=False),
)
transform_task = PythonOperator(
task_id="transform",
python_callable=transform,
op_args=[],
op_kwargs=dict(a2=1,b2='bb',c2=False),
)
load_task = PythonOperator(
task_id="load",
python_callable=load,
op_args=[],
op_kwargs=dict(a3=1,b3='bb',c3=False),
)
extract_task >> transform_task >> load_task
