任务调度管理工具：Airflow

本是古典何须时尚 2023-01-15 14:23 190阅读 0赞

# 安装 #

airflow是由python编写的，安装也比较简单，可以直接通过pip命令安装：

`pip install airflow`

pip install airflow
    
    # 安装airflow使用mysql的相关依赖
    pip install airflow[mysql]
    
    # 可以用all安装所有相关的插件依赖
    pip install airflow[all]

但是，官方提到，这样安装容易缺失部分前置依赖，CONSTRAINT\_URL里面包含了所有airflow涉及到的依赖；

我们可以在pip install时加上–constraint，指定安装需要的所有前置依赖。

AIRFLOW_VERSION=1.10.15
    
    PYTHON_VERSION=3.6
    
    CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"
    
    pip install "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}"

# 基本概念 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NneXVhbnNoaQ_size_16_color_FFFFFF_t_70]

## operation ##

operation即具体的执行操作，airflow支持**BashOperator**：shell脚本命令的执行操作、**PythonOperator**：python脚本、**EmailOperator**：邮箱操作等等

## task ##

一个task代表一个任务，是airflow的执行单位，如上图中的write\_a即是一个task，对应一个operation。

## DAG ##

DAG即是由许多个task组成的有向无环图。如上图由write\_a、write\_b、write\_c三个task组成的dag。

# 配置 #

## 生成配置文件 ##

airflow安装之后是不会立即生成相关目录的，所以，当然也不会有配置文件。

只要我们执行任意一条airflow的命令之后，airflow就会生成其对应的目录`~/airflow`，里面会有`airflow.cfg`配置文件。

# 查看airflow的版本
    airflow version
    
    # 查看airflow的帮助说明
    airflow --help

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NneXVhbnNoaQ_size_16_color_FFFFFF_t_70 1]

## 数据库配置 ##

airflow的调度需要依赖数据库来实现。

airflow默认自带了sqlite数据库，如果想使用mysql，则需要在airflow.cfg中修改：

sql_alchemy_conn = mysql://username:password@localhost:3306/airflow

## executor配置 ##

其中，还有一个关键的配置项：`executor`

SequentialExecutor：顺序执行的executor，sqllite数据库下，只能选择这个。意味着同个时间只能执行一个task，即不支持并发。

LocalExecutor：执行并发。并发度由`parallelism = 32`决定。

CeleryExecutor：分布式结构。即可以支持多个worker。

DaskExecutor、KubernetesExecutor：即可以将task放在Dask、Kubernetes环境下执行。

# 启动 #

## 前端UI ##

airflow webserver -p 8080 -D

\-D表示以守护线程启动，默认的端口为8080

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NneXVhbnNoaQ_size_16_color_FFFFFF_t_70 2]

## scheduler ##

airflow scheduler -D

# 添加DAG #

## dag的py脚本 ##

from datetime import datetime, timedelta
    
    from airflow import DAG
    from airflow.operators.bash_operator import BashOperator
    
    default_args = { 
        'owner': 'airflow',
        'depends_on_past': False,
        'start_date': datetime.now(),
        'email': ['airflow@example.com'],
        'email_on_failure': True,
        'email_on_retry': False,
        'retries': 1,
        'retry_delay': timedelta(minutes=5),
        # 'queue': 'bash_queue',
        # 'pool': 'backfill',
        # 'priority_weight': 10,
        # 'end_date': datetime(2016, 1, 1),
        # 'wait_for_downstream': False,
        # 'dag': dag,
        # 'sla': timedelta(hours=2),
        # 'execution_timeout': timedelta(seconds=300),
        # 'on_failure_callback': some_function,
        # 'on_success_callback': some_other_function,
        # 'on_retry_callback': another_function,
        # 'sla_miss_callback': yet_another_function,
        # 'trigger_rule': 'all_success'
    }
    
    dag = DAG(
        'my_first_dag',
        default_args=default_args,
        description='A simple tutorial DAG',
        schedule_interval=timedelta(minutes=1),
    )
    
    t1 = BashOperator(
        task_id='write_a',
        bash_command='python /Users/hong/Desktop/Develop/my_airflow/operations/first_operation.py a',
        dag=dag,
    )
    
    t2 = BashOperator(
        task_id='write_b',
        bash_command='python /Users/hong/Desktop/Develop/my_airflow/operations/first_operation.py b',
        dag=dag,
    )
    
    t3 = BashOperator(
        task_id='write_c',
        bash_command='python /Users/hong/Desktop/Develop/my_airflow/operations/first_operation.py c',
        dag=dag,
    )
    
    t1.set_downstream(t2)
    t1.set_downstream(t3)

我们按照以上的格式，将脚本文件如first\_dag.py，放置到airflow的dag目录下，airflow会自动将加载。

根据airflow.cfg的配置项：`dags_folder = /Users/hong/airflow/dags`

## 开启dag ##

然后我们在airflow的前端界面`localhost:8080`就可以看到相关的dag任务了，点击**Off**按钮，将其转化为**On**，它就会开始安排调度定时任务了。

![在这里插入图片描述][20210422215517223.png]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NneXVhbnNoaQ_size_16_color_FFFFFF_t_70 3]

# 日志查看 #

在前端界面下，Browse --> Task Instances可以看到，每个dag的所有task运行记录。

在Browse --> Dag Runs可以看到，所有dag任务的执行记录。

我们点进去之后，每个task会有对应的运行状态，然后单独选择某个task单击，选择View Log就可以看到运行日志了。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NneXVhbnNoaQ_size_16_color_FFFFFF_t_70 4]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NneXVhbnNoaQ_size_16_color_FFFFFF_t_70 5]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NneXVhbnNoaQ_size_16_color_FFFFFF_t_70 6]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NneXVhbnNoaQ_size_16_color_FFFFFF_t_70]: /images/20221022/e2a70d7f22b44a02b9bd36e5cf8555b6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NneXVhbnNoaQ_size_16_color_FFFFFF_t_70 1]: /images/20221022/7d1ca680e6ff4ec1aaf303bef48bc2f7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NneXVhbnNoaQ_size_16_color_FFFFFF_t_70 2]: /images/20221022/c8cc24a51574413882118cfd18a3ac40.png
[20210422215517223.png]: /images/20221022/7dad2b2e2f264d2194fddb0d206d7b13.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NneXVhbnNoaQ_size_16_color_FFFFFF_t_70 3]: /images/20221022/02347f80a2984e529a0889383b36333e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NneXVhbnNoaQ_size_16_color_FFFFFF_t_70 4]: /images/20221022/966bf4067a7444dd9b58296a66f527c3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NneXVhbnNoaQ_size_16_color_FFFFFF_t_70 5]: /images/20221022/a498149226524405bfa38e0b7a0e30f8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NneXVhbnNoaQ_size_16_color_FFFFFF_t_70 6]: /images/20221022/6985f8a947be435ca79875ff9fd4fdc9.png