Airflow 是什么

为什么使用 Airflow

  • 智能调度
  • 图形化展示任务关系
  • 程序化流水线定义
  • 流水线间数据交互
  • 扩展性强
  • 分布式，可靠性高
  • 执行方式多样化，除了定时执行还可手动触发，api 触发等

Airflow 常用的场景

Airflow 主要用于执行预定的批处理作业。它能够很好地管理不同的批量作业的关系，并给将复杂的关系图形化展示。

  • 系统或运用的日常维护任务，批量作业
  • CD 部署任务，灰度发布，蓝绿部署等
  • 数据分析，数据流管道管理

一句话：任何批量任务或者需要手动去触发执行的任务都可以考虑一下 Airflow

Airflow 安装

airlow 使用 ansible role 方式安装 https://github.com/idealista/airflow-role 0. 安装准备软件 ansible、gcc、python-devel

shell
1
$ yum install ansible gcc python-devel -y
  1. 创建文件requirements.yml
plaintext
1
2
3
- src: idealista.airflow-role
  version: 1.0.0
  name: airflow
  1. 下载 airflow role
shell
1
$ ansible-galaxy install -p roles -r requirements.yml -f
  1. 更改 role 中的 install.yml
    将 roles/airflow/tasks/install.yml 中的apt模块改成package模块
    将 roles/airflow/tasks/install.yml 中的 Copy Daemon scripts 中的 when 条件去掉
shell
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
$ cat roles/airflow/tasks/install.yml
...
- name: Airflow | Installing dependencies
  package:
    name: "{{ item }}"
    state: present
  with_items: "{{ airflow_required_libs }}"
...
- name: Airflow | Copy Daemon scripts
  template:
    src: "{{ item.key }}.service.j2"
    dest: /lib/systemd/system/{{ item.key }}.service
    mode: 0644
  notify: restart {{ item.key }}
  with_dict: "{{ airflow_services }}"

更改 role 下的 config.yml 及 templates/airflow-***.service.j2 文件
将/usr/local/bin/airflow改为/usr/bin/airflow

shell
1
2
3
4
5
6
7
8
9
$ cat roles/airflow/tasks/config.yml
...
- name: Airflow | Initializing DB
  shell: AIRFLOW_HOME={{ airflow_home }} airflow initdb
...
$ cat roles/airflow/templates/airflow-webserver.service.j2
...
ExecStart=/usr/bin/airflow webserver --pid /run/airflow/webserver.pid
...

目的该 role 使用 pkg 只能在 Debian  或 Ubuntu 环境下安装，更改后，可以在 centos 环境下安装 4. 更新 ansible 的 hosts 文件

shell
1
2
$ cat /etc/ansible/hosts
127.0.0.1 ansible_connection=local
  1. 创建部署 airflow 的 ansible yaml
shell
1
2
3
4
5
$ cat airflow.yml
---
- hosts: all
  roles:
  - { role: airflow }
  1. 安装 airflow
shell
1
$ ansible-playbook airflow.yml

执行完以上操作后本机就安装好了 airflow，同时也配置好了 service 等。 7. 手动初始化 db

shell
1
$ HOME=/etc/airflow; airflow initdb
  1. 更新配置/etc/airflow/airflow.cfg
shell
1
2
3
4
5
$ cat /etc/airflow/airflow.cfg
...
max_threads = 1
dagbag_import_timeout = 30
...
  1. 启动 airflow-worker
shell
1
$ systemctl restart airflow-worker

安装好后展示

Airflow主页

问题记录

  • 如果使用 CeleryExecutor，可以安装 Celery 管理器 flower。，默认会安装 tornado 最新版本，需要限定 tornado 版本需限制在 4.2.0。安装完成后运行AIRFLOW_HOME=/etc/airflow airflow flower
  • 使用 Mysql 作为后台数据库时，安装 mysql 扩展请安装 mysql-python。
    同时 broker_url 设置为broker_url = mysql://airflow:airflow@localhost:3306/airflow
  • 测试 Airflow 中的 DAG 时，需要在界面上将它先把状态改为 On 后，再点击运行。

相关文章

Airflow 中文文档
任务调度神器 airflow 之初体验
airflow 安装，部署，填坑
airflow 配置 CeleryExecutor
浅谈调度工具——Airflow
如何部署一个健壮的 apache-airflow 调度系统
Airflow 工作模式及适用场景
浅谈调度工具——Airflow