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
大规模运行 Apache 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