Airflow安装及使用
文章目录
Airflow基本概念
概述
Airflow是一个以编程方式编写，安排和监视工作流的平台。
使用Airflow将工作流编写任务的有向无环图（DAG）。Airflow计划程序在遵循指定的依赖项，同时在一组工作线程上执行任务。丰富的命令实用程序使在DAG上执行复杂的调度变的轻而易举。丰富的用户界面使查看生产中正在运行的管道，监视进度以及需要时对问题进行故障排除变的容易。
名词
（1）Dynamic：Airflow配置需要实用Python，允许动态生产管道。这允许编写可动态。这允许编写可动态实例化管道的代码。
（2）Extensible：轻松定义自己的运算符，执行程序并扩展库，使其适合于您的环境。
（3）Elegant：Airlfow是精简的，使用功能强大的Jinja模板引擎，将脚本参数化内置于Airflow的核心中。
（4）Scalable：Airflow具有模板块架构，并使用消息队列来安排任意数量的工作任务。
Airflow安装
安装Python环境
Superset是由Python语言编写的Web应用，要求Python3.8的环境。
安装Miniconda
conda是一个开源的包、环境管理器，可以用于在同一个机器上安装不同Python版本的软件包及其依赖，并能够在不同的Python环境之间切换，Anaconda包括Conda、Python以及一大堆安装好的工具包，比如：numpy、pandas等，Miniconda包括Conda、Python。
此处，我们不需要如此多的工具包，故选择MiniConda。
（1）下载Miniconda（Python3版本）
下载地址：https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh
（2）安装Miniconda
执行以下命令进行安装，并按照提示操作，直到安装完成。
[atguigu@hadoop102 lib]$ bash Miniconda3-latest-Linux-x86_64.sh
- 1
在安装过程中，出现以下提示时，可以指定安装路径：
（3）加载环境变量配置文件，使之生效
[atguigu@hadoop102 lib]$ source ~/.bashrc
- 1
（4）取消激活base环境
Miniconda安装完成后，每次打开终端都会激活其默认的base环境，我们可通过以下命令，禁止激活默认base环境。
[atguigu@hadoop102 lib]$ conda config --set auto_activate_base false
- 1
创建Python3.8环境
（1）配置conda国内镜像
(base) [atguigu@hadoop102 ~]$ conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free
(base) [atguigu@hadoop102 ~]$ conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main
(base) [atguigu@hadoop102 ~]$ conda config --set show_channel_urls yes
- 1
- 2
- 3
（2）创建python3.8环境
(base) [atguigu@hadoop102 ~]$ conda create --name airflow python=3.8
- 1
说明：conda环境管理常用命令：
- 创建环境：conda create -n env_name
- 查看所有环境：conda info --envs
- 删除一个环境：conda remove -n env_name --all
（3）激活airflow环境
(base) [atguigu@hadoop102 ~]$ conda activate airflow
- 1
说明：退出当前环境。
(superset) [atguigu@hadoop102 ~]$ conda deactivate
- 1
（4）执行python -V命令查看python版本
(airflow) [atguigu@hadoop102 software]$ python -V
Python 3.8.13
- 1
- 2
安装Airflow
（1）更改pip的源
[atguigu@hadoop102 software]$ conda activate airflow
(airflow) [atguigu@hadoop102 software]$ pip install numpy -i https://pypi.tuna.tsinghua.edu.cn/simple
(airflow) [atguigu@hadoop102 software]$ sudo mkdir ~/.pip
(airflow) [atguigu@hadoop102 software]$ sudo vim ~/.pip/pip.conf
#添加以下内容
[global]
index-url = https://pypi.tuna.tsinghua.edu.cn/simple
[install]
trusted-host = https://pypi.tuna.tsinghua.edu.cn
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
（2）安装airflow
(airflow) [atguigu@hadoop102 software]$ pip install "apache-airflow==2.4.3"
- 1
（3）初始化airflow
(airflow) [atguigu@hadoop102 software]$ airflow db init
- 1
（4）查看版本
(airflow) [atguigu@hadoop102 software]$ airflow version
2.4.3
- 1
- 2
（5）airflow安装好存放路径
(airflow) [atguigu@hadoop102 airflow]$ pwd
/home/atguigu/airflow
- 1
- 2
（6）启动airflow web服务,启动后浏览器访问http://hadoop102:8080
(airflow) [atguigu@hadoop102 airflow]$ airflow webserver -p 8080 -D
- 1
（7）启动airflow调度
(airflow) [atguigu@hadoop102 airflow]$ airflow scheduler -D
- 1
（8）创建账号
(airflow) [atguigu@hadoop102 airflow]$ airflow users create \
> --username admin \
> --firstname atguigu \
> --lastname atguigu \
> --role Admin \
> --email yhm@atguigu.com
- 1
- 2
- 3
- 4
- 5
- 6
（9）启动停止脚本
[atguigu@hadoop102 bin]$ vim af.sh
#!/bin/bash
case $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
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
修改数据库为MySQL
（1）在MySQL中建库
mysql> CREATE DATABASE airflow_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
- 1
（2）如果报错Linux error:1425F102:SSL routines:ssl_choose_client_version:unsupported protocol，可以关闭MySQL的SSL证书
查看SSL是否开启 YES为开启
mysql> SHOW VARIABLES LIKE '%ssl%';
+---------------+-----------------+
| Variable_name | Value |
+---------------+-----------------+
| have_openssl | YES |
| have_ssl | YES |
| ssl_ca | ca.pem |
| ssl_capath | |
| ssl_cert | server-cert.pem |
| ssl_cipher | |
| ssl_crl | |
| ssl_crlpath | |
| ssl_key | server-key.pem |
+---------------+-----------------+
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
（3）修改配置文件my.cnf，加入以下内容：
# disable_ssl
skip_ssl
- 1
- 2
（4）添加python连接的依赖：
官网介绍的方法有两种，这里我们选择下面的连接器。
(airflow) [atguigu@hadoop102 airflow]$ pip install mysql-connector-python
- 1
（5）修改airflow的配置文件：
[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:123456@hadoop102:3306/airflow_db
- 1
- 2
- 3
- 4
- 5
- 6
- 7
（6）关闭airflow，初始化后重启：
(airflow) [atguigu@hadoop102 ~]$ af.sh stop
(airflow) [atguigu@hadoop102 airflow]$ airflow db init
(airflow) [atguigu@hadoop102 ~]$ af.sh start
- 1
- 2
- 3
（7）初始化报错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'
- 1
重启MySQL会造成参数失效，推荐将参数写入到配置文件my.cnf中。
sql_mode = STRICT_TRANS_TABLES,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION
- 1
（8）重新创建账号登录：
(airflow) [atguigu@hadoop102 airflow]$ airflow users create \
--username admin \
--firstname atguigu \
--lastname atguigu \
--role Admin \
--email yhm@atguigu.com
- 1
- 2
- 3
- 4
- 5
- 6
修改执行器
官网不推荐在开发中使用顺序执行器，会造成任务调度阻塞。
修改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
- 1
- 2
- 3
- 4
- 5
- 6
可以使用官方推荐的几种执行器，也可以自定义。这里我们选择本地执行器即可。
部署使用
（1）测试环境启动
本次测试使用的是spark的官方案例，所有需要启动hadoop和spark的历史服务器。
[atguigu@hadoop102 bin]$ myhadoop.sh start
[atguigu@hadoop102 bin]$ cd /opt/module/spark-yarn/sbin/start-history-server.sh
- 1
- 2
（2）查看Airflow配置文件
(python3) [root@airflow work-py]# vim ~/airflow/airflow.cfg
- 1
（3）编写.py脚本，创建work-py目录用于存放python调度脚本
(airflow) [atguigu@hadoop102 airflow]$ mkdir ~/airflow/dags
(airflow) [atguigu@hadoop102 airflow]$ cd dags/
(airflow) [atguigu@hadoop102 dags]$ vim test.py
#!/usr/bin/python
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
default_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(
# 任务id
task_id=‘dwd’,
# 任务命令
bash_command=‘ssh hadoop102 "/opt/module/spark-yarn/bin/spark-submit --class org.apache.spark.examples.SparkPi --master yarn /opt/module/spark-yarn/examples/jars/spark-examples_2.12-3.1.3.jar 10 "’,
# 重试次数
retries=3,
# 把任务添加进图中
dag=dag)
t2 = BashOperator(
task_id=‘dws’,
bash_command=‘ssh hadoop102 "/opt/module/spark-yarn/bin/spark-submit --class org.apache.spark.examples.SparkPi --master yarn /opt/module/spark-yarn/examples/jars/spark-examples_2.12-3.1.3.jar 10 "’,
retries=3,
dag=dag)
t3 = BashOperator(
task_id=‘ads’,
bash_command=‘ssh hadoop102 "/opt/module/spark-yarn/bin/spark-submit --class org.apache.spark.examples.SparkPi --master yarn /opt/module/spark-yarn/examples/jars/spark-examples_2.12-3.1.3.jar 10 "’,
retries=3,
dag=dag)
<span class="token comment"># 设置任务依赖</span>
t2.set_upstream(t1)
t3.set_upstream(t2)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
-
必须导包
from airflow import DAG from airflow.operators.bash_operator import BashOperator- 1
- 2
-
default_args 设置默认参数。
-
depends_on_past 是否开启任务依赖。
-
schedule_interval 调度频率。
-
retries 重试次数 。
-
start_date 开始时间。
-
BashOperator 具体执行任务，如果为true前置任务必须成功完成才会走下一个依赖任务，如果为false则忽略是否成功完成。
-
task_id 任务唯一标识（必填）。
-
bash_command 具体任务执行命令。
-
set_upstream 设置依赖 如上图所示ads任务依赖dws任务依赖dwd任务。
（4）等待一段时间，刷新任务列表
(airflow) [atguigu@hadoop102 airflow]$ airflow dags list
-------------------------------------------------------------------
DAGS
-------------------------------------------------------------------
example_bash_operator
example_branch_dop_operator_v3
example_branch_operator
example_complex
example_external_task_marker_child
example_external_task_marker_parent
example_http_operator
example_kubernetes_executor_config
example_nested_branch_dag
example_passing_params_via_test_command
example_pig_operator
example_python_operator
example_short_circuit_operator
example_skip_dag
example_subdag_operator
example_subdag_operator.section-1
example_subdag_operator.section-2
example_trigger_controller_dag
example_trigger_target_dag
example_xcom
latest_only
latest_only_with_trigger
test
test_utils
tutorial
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
（5）已出现test任务，刷新页面
（6）点击运行
（7）点击成功任务，查看日志
（8）查看dag图、甘特图
（9）查看脚本代码
Dag任务操作
删除Dag任务
主要删除DAG任务不会删除底层文件，过一会还会自动加载回来。
查看当前所有dag任务
# 查看所有任务
(airflow) [atguigu@hadoop102 airflow]$ airflow list_dags
# 查看单个任务
(airflow) [atguigu@hadoop102 airflow]$ airflow tasks list test --tree
- 1
- 2
- 3
- 4
配置邮件服务器
（1）保证邮箱已开SMTP服务
（2）修改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
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
（3）重启airflow
[atguigu@hadoop102 bin]$ af.sh stop
[atguigu@hadoop102 bin]$ af.sh start
- 1
- 2
（4）编辑test.py脚本，并且替换
[root@airflow ~]# cd /opt/module/work-py/
[root@airflow work-py]# vim test.py
- 1
- 2
#!/usr/bin/python
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.email_operator import EmailOperator
from datetime import datetime, timedelta
default_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(
# 任务id
task_id=‘dwd’,
# 任务命令
bash_command=‘ssh hadoop102 "/opt/module/spark-yarn/bin/spark-submit --class org.apache.spark.examples.SparkPi --master yarn /opt/module/spark-yarn/examples/jars/spark-examples_2.12-3.1.3.jar 10 "’,
# 重试次数
retries=3,
# 把任务添加进图中
dag=dag)
t2 = BashOperator(
task_id=‘dws’,
bash_command=‘ssh hadoop102 "/opt/module/spark-yarn/bin/spark-submit --class org.apache.spark.examples.SparkPi --master yarn /opt/module/spark-yarn/examples/jars/spark-examples_2.12-3.1.3.jar 10 "’,
retries=3,
dag=dag)
t3 = BashOperator(
task_id=‘ads’,
bash_command=‘ssh hadoop102 "/opt/module/spark-yarn/bin/spark-submit --class org.apache.spark.examples.SparkPi --master yarn /opt/module/spark-yarn/examples/jars/spark-examples_2.12-3.1.3.jar 10 "’,
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)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
（5）查看页面是否生效
（6）运行测试
