Airflow学习笔记

1 概述

Airflow是一个以编程方式编写（要用python脚本），安排和监视工作流的平台。使用Airflow将工作流编写任务的有向无环图（DAG）。Airflow计划程序在遵循指定的依赖项，同时在一组工作线程上执行任务。丰富的命令实用程序使在DAG上执行复杂的调度变的轻而易举。丰富的用户界面使查看生产中正在运行的管道，监视进度以及需要时对问题进行故障排除变的容易。
相关概念：
1. Dynamic：Airflow配置需要实用Python，允许动态生产管道。这允许编写可动态实例化管道的代码。
2. Extensible：轻松定义自己的运算符，执行程序并扩展库，使其适合于您的环境。
3. Elegant：Airlfow是精简的，使用功能强大的Jinja模板引擎，将脚本参数化内置于Airflow的核心中。
4. Scalable：Airflow具有模板块架构，并使用消息队列来安排任意数量的工作任务。

2 安装

airflow官网：https://airflow.apache.org
airflow需要python3.8的环境，这里安装miniconda来创建python环境

具体安装步骤：

在conda下创建python3.8环境，命名为airflow

切换airflow环境，更改pip源

sudo mkdir ~/.pip
sudo vim  ~/.pip/pip.conf
#添加以下内容
[global]
index-url = https://pypi.tuna.tsinghua.edu.cn/simple
[install]
trusted-host = https://pypi.tuna.tsinghua.edu.cn

安装airflow包
```
pip install apache-airflow==2.4.3
```
初始化aiflow
```
airflow db init
```
启动airflow服务
```
airflow webserver -p 8080 -D
```
启动airflow调度
```
airflow scheduler -D
```

创建账号，创建后会要求输入密码

airflow users create \--username admin \--firstname F \--lastname F \--role Admin \--email F@F.com

airflow启停脚本

#!/bin/bashcase $1 in
"start"){echo " --------启动 airflow-------"ssh hadoop102 "conda activate airflow;airflow webserver -p 8080 -D;airflow scheduler -D; conda deactivate"
};;
"stop"){echo " --------关闭 airflow-------"ps -ef|egrep 'scheduler|airflow-webserver'|grep -v grep|awk '{print $2}'|xargs kill -15 
};;
esac

2.1 修改默认元数据库

修改元数据库为mysql，默认使用的是本地的sqlite，效果不好。（P8）

在mysql中创建库

mysql> CREATE DATABASE airflow_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

安装python连接mysql依赖
```
pip install mysql-connector-python
```

修改airflow的配置文件

vim ~/airflow/airflow.cfg[database]
# The SqlAlchemy connection string to the metadata database.
# SqlAlchemy supports many different database engines.
# More information here:
# http://airflow.apache.org/docs/apache-airflow/stable/howto/set-up-database.html#database-uri
#sql_alchemy_conn = sqlite:home/atguigu/airflow/airflow.db
sql_alchemy_conn = mysql+mysqlconnector://root:root@hadoop102:3306/airflow_db

关闭airflow，初始化后重启
```
(airflow) [atguigu@hadoop102 ~]$ af.sh stop
(airflow) [atguigu@hadoop102 airflow]$ airflow db init
(airflow) [atguigu@hadoop102 ~]$ af.sh start
```
若初始化报错1067 - Invalid default value for ‘update_at’，原因：字段 ‘update_at’ 为 timestamp类型，取值范围是：1970-01-01 00:00:00 到 2037-12-31 23:59:59（UTC +8 北京时间从1970-01-01 08:00:00 开始），而这里默认给了空值，所以导致失败。推荐修改mysql存储时间戳格式：
```
mysql> set GLOBAL sql_mode ='STRICT_TRANS_TABLES,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION'
```
重启MySQL会造成参数失效，推荐将参数写入到配置文件/etc/my.cnf中并重启，然后再一次初始化
```
sql_mode = STRICT_TRANS_TABLES,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTIONsudo systemctl restart mysqld
```

重新创建账号登录

airflow users create \--username admin \--firstname F \--lastname F \--role Admin \--email F@F.com

2.2 修改默认执行器

官网不推荐在开发中使用顺序执行器SequentialExecutor，类似单进单出的队列，会造成任务调度阻塞。

修改airflow的配置文件，可以使用官方推荐的几种执行器，也可以自定义。这里我们选择本地执行器即可。

[core]
# The executor class that airflow should use. Choices include
# ``SequentialExecutor``, ``LocalExecutor``, ``CeleryExecutor``, ``DaskExecutor``,
# ``KubernetesExecutor``, ``CeleryKubernetesExecutor`` or the
# full import path to the class when using a custom executor.
executor = LocalExecutor

3 使用

默认的存放airflow调度脚本的目录在~/airflow/dags，我们只需要在这个目录下编写调度脚本即可。

3.1 添加DAG

在dags目录创建python脚本，对一些关键信息进行修改

#!/usr/bin/python
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedeltadefault_args = {# 用户，任写'owner': 'test_owner', # 是否开启任务依赖'depends_on_past': True, # 邮箱，任写'email': ['403627000@qq.com'],# 启动时间，任写'start_date':datetime(2022,11,28),# 出错是否发邮件报警'email_on_failure': False,# 重试是否发邮件报警'email_on_retry': False,# 重试次数'retries': 1,# 重试时间间隔'retry_delay': timedelta(minutes=5),
}
# 声明任务图，传递参数，设置调度时间间隔
dag = DAG('test', default_args=default_args, schedule_interval=timedelta(days=1))# 创建单个任务
t1 = BashOperator(# 任务idtask_id='dwd',# 任务命令bash_command='echo 1',# 重试次数retries=3,# 把任务添加进图中dag=dag)t2 = BashOperator(task_id='dws',bash_command='echo 2',retries=3,dag=dag)t3 = BashOperator(task_id='ads',bash_command='echo 3',retries=3,dag=dag)# 设置任务依赖
t2.set_upstream(t1)
t3.set_upstream(t2)

等待一段时间，刷新任务列表，可以在web页面查看，也可以通过以下命令查看
```
airflow dags list
```
在web页面上开启任务调度脚本，点进去能到查看DAG，以及DAG中每个任务的执行情况，执行日志等

3.2 删除DAG

在页面上点击删除按钮删除DAG任务，会连同任务的执行记录一块删除
删除底层的python脚本，否则过段时间会自动重新加载这个任务

3.3 修改DAG

airflow支持动态修改，直接修改python脚本即可

3.4 查看DAG

3.5 配置邮件服务器

保证邮箱已开SMTP服务

airflow配置文件，用stmps服务对应587端口

(airflow) [atguigu@hadoop102 airflow]$ vim ~/airflow/airflow.cfg  
smtp_host = smtp.qq.com
smtp_starttls = True
smtp_ssl = False
smtp_user = 403627000@qq.com
# smtp_user =
smtp_password = qluxdbuhgrhgbigi
# smtp_password =
smtp_port = 587
smtp_mail_from = 403627000@qq.com

重启airflow

python脚本也需要有所更改

#!/usr/bin/python
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedeltadefault_args = {# 用户，任写'owner': 'test_owner', # 是否开启任务依赖'depends_on_past': True, # 邮箱，任写'email': ['403627000@qq.com'],# 启动时间，任写'start_date':datetime(2022,11,28),# 出错是否发邮件报警'email_on_failure': False,# 重试是否发邮件报警'email_on_retry': False,# 重试次数'retries': 1,# 重试时间间隔'retry_delay': timedelta(minutes=5),
}
# 声明任务图，传递参数，设置调度时间间隔
dag = DAG('test', default_args=default_args, schedule_interval=timedelta(days=1))# 创建单个任务
t1 = BashOperator(# 任务idtask_id='dwd',# 任务命令bash_command='echo 1',# 重试次数retries=3,# 把任务添加进图中dag=dag)t2 = BashOperator(task_id='dws',bash_command='echo 2',retries=3,dag=dag)t3 = BashOperator(task_id='ads',bash_command='echo 3',retries=3,dag=dag)email=EmailOperator(task_id="email",to="yaohm163@163.com ",subject="test-subject",html_content="<h1>test-content</h1>",cc="403627000@qq.com ",dag=dag)# 设置任务依赖
t2.set_upstream(t1)
t3.set_upstream(t2)
email.set_upstream(t3)