airflow配置信息

Airflow 1.10+安裝

本次安裝Airflow版本為1.10+，其需要依賴Python和DB，本次選擇的DB為Mysql。
本次安裝組件及版本如下：Airflow == 1.10.0
Python == 3.6.5
Mysql == 5.7

整體流程

1. 建表
2. 安裝
3. 配置
4. 運行
5. 配置任務

啟動schedule
airflow scheduler -D
啟動webserver
airflow webserver -D

ps -ef|grep -Ei '(airflow-webserver)'| grep master | awk '{print $2}'|xargs -i kill {}
ps -ef | grep -Ei 'airflow' | grep -v 'grep' | awk '{print $2}' | xargs -i kill {}

建庫、建用戶

庫名為airflow
'create database airflow;'
建用戶

用戶名為airflow，並且設置所有ip均可以訪問。

create user 'airflow'@'%' identified by 'airflow';
create user 'airflow'@'localhost' identified by 'airflow';

用戶授權
這里為新建的airflow用戶授予airflow庫的所有權限

grant all on airflow.* to 'airflow'@'%';
flush privileges

Airflow安裝

這里通過 virtualenv 進行安裝。

----- 通過virtualenv安裝

$ mkdir /usr/local/virtual_env && cd /usr/local/virtual_env # 創建目錄
$ virtualenv --no-site-packages airflow --python=python # 創建虛擬環境
$ source /usr/local/virtual_env/airflow/bin/activate # 激活虛擬環境

----- 安裝指定版本或者默認
$ pip install apache-airflow -i https://pypi.douban.com/simple
在安裝完一堆的依賴后，就需要配置 AIRFLOW_HOME 環境變量，后續的 DAG 和 Plugin 都將以該目錄作為根目錄查找，如上，可以直接設置為 /tmp/project 。

報錯
ERROR: flask 1.1.1 has requirement Jinja2>=2.10.1, but you'll have jinja2 2.10 which is incompatible.
ERROR: flask 1.1.1 has requirement Werkzeug>=0.15, but you'll have werkzeug 0.14.1 which is incompatible.

執行：pip3 install -U Flask1.0.4
執行：pip3 install -U pika0.13.1

重新執行 ：pip install apache-airflow -i https://pypi.douban.com/simple

----- 設置環境變量
(airflow) $ export AIRFLOW_HOME=/tmp/airflow

----- 查看其版本信息
(airflow) $ airflow version

---

____ |( )_______ / /________ __
____ /| |_ /__ / / __ / __ _ | /| / /
___ ___ | / _ / _ / _ / / // / |/ |/ /
// |// // // // _/____/|__/
v1.8.0
執行了上述的命令后，會生成 airflow.cfg 和 unittests.cfg 兩個文件，其中前者是一個配置文件 。

airflow 配置

----- 修改Airflow DB配置

1. 安裝Mysql模塊

pip install "apache-airflow[mysql]"
這里可以簡單說下，airflow依賴的其他組件均可以此方式安裝。在之后安裝password組件同樣是通過此方式。

修改Airflow DB配置
修改${AIRFLOW_HOME}/airflow.cfg

sql_alchemy_conn = mysql+mysqldb://airflow:airflow@localhost:3306/airflow
參數的格式為mysql://帳號:密碼@ip:port/db

初始化db
新建airflow依賴的表。

airflow initdb

如報錯 Can't connect to local MySQL server through socket '/var/lib/mysql/mysql.sock' (2)
需改sql_alchemy_conn = mysql+mysqldb://airflow:airflow@127.0.0.1:3306/airflow

2. 用戶認證

本文采用的用戶認證方式為password方式，其他方式如LDAP同樣支持但是本文不會介紹。筆者在安裝時實驗過LDAP方式但是未成功過。

安裝passsword組件
pip install "apache-airflow[password]"
2. 修改 airflow.cfg

[webserver]
authenticate = True
auth_backend = airflow.contrib.auth.backends.password_auth
3. 在python環境中執行如下代碼以添加賬戶：

import airflow  
from airflow import models, settings  
from airflow.contrib.auth.backends.password_auth import PasswordUser  
user = PasswordUser(models.User())  
user.username = 'admin'  # 用戶名
user.email = '[e-mail]' # 用戶郵箱  
user.password = 'password'   # 用戶密碼
session = settings.Session()  
session.add(user)  
session.commit()  
session.close()  
exit() 

3. 配置郵件服務

此配置設置的是dag的task失敗或者重試時發送郵件的發送者。配置如下：

[smtp]
# If you want airflow to send emails on retries, failure, and you want to use
# the airflow.utils.email.send_email_smtp function, you have to configure an
smtp_host = smtp.163.com
smtp_starttls = True
smtp_ssl = False
# Uncomment and set the user/pass settings if you want to use SMTP AUTH
smtp_user = [e-mail]
smtp_password = password
smtp_port = 25
smtp_mail_from = [e-mail]
接下來簡單把dag的Python代碼列出來，以供參考：

default_args = {
  'owner': 'ownerExample',
  'start_date': datetime(2018, 9, 18),
  'email': ['[e-mail]'], # 出問題時，發送報警Email的地址，可以填多個，用逗號隔開。
  'email_on_failure': ['[e-mail]'], # 任務失敗且重試次數用完時發送Email。
  'email_on_retry': True, # 任務重試時是否發送Email
  'depends_on_past': False, # 是否依賴於過去。如果為True，那么必須要昨天的DAG執行成功了，今天的DAG才能執行。
  'retries': 3,
  'retry_delay': timedelta(minutes=3),
}

4、配置Executor

設置Executor
修改：airflow.cfg

executor = LocalExecutor
本文中由於只有單節點所以使用的是LocalExecutor模式。

5. 修改log地址

[core]
# dags存放路徑
dags_folder = /home/xxx/airflow/dags
# 獲取服務器IP的方式
hostname_callable = socket.getfqdn
# 時區，可以是UTC(默認)，也可以換成國內Asia/Shanghai
default_timezone = utc
# airflow支持並行性的工作器，有`SequentialExecutor`(默認，順序執行), `LocalExecutor`(本地執行), `CeleryExecutor`(遠程執行), `DaskExecutor`
executor = SequentialExecutor
# 數據庫連接設置
sql_alchemy_conn = sqlite:////home/xxx/airflow/airflow.db
# 數據庫編碼方式
sql_engine_encoding = utf-8
# 是否與SqlAlchemy庫進行數據交互
sql_alchemy_pool_enabled = True
# 最大數據庫連接數
sql_alchemy_pool_size = 5
# 控制每個Airflow worker可以同時運行task實例的數量
parallelism = 32
# 用來控制每個dag運行過程中最大可同時運行的task實例數，若DAG中沒有設置concurrency，則使用默認值
dag_concurrency = 16
# 創建新的DAG時，是否暫停
dags_are_paused_at_creation = True
# 同一時間最大運行dag的數量，默認為16
max_active_runs_per_dag = 16
# 加載示例dags，默認為True
load_examples = True

[scheduler]
child_process_log_directory = servers/logs/airflow/scheduler

6. 修改webserver地址

修改webserver地址
[webserver]
base_url = http://host:port
可以通過上面配置的地址訪問webserver。

7. 可選配置

（可選）修改Scheduler線程數
如果調度任務不多的話可以把線程數調小，默認為32。參數為：parallelism


（可選）不加載example dag
如果不想加載示例dag可以把load_examples配置改為False，默認為True。這個配置只有在第一次啟動airflow之前設置才有效。


如果此方法不生效，可以刪除${PYTHON_HOME}/site-packages/airflow/example_dags目錄，也是同樣的效果。

（可選）修改檢測新dag間隔
修改min_file_process_interval參數為10，每10s識別一次新的dag。默認為0，沒有時間間隔。

8.celery配置信息

#  配置celery的broker_url(存儲要執行的命令然后celery的worker去消費)
broker_url = redis://redis:6379/0
# 配置celery的result_backend(存儲任務執行狀態)、 也可以用redis存儲
result_backend = db+postgresql://postgres:airflow@postgres/airflow

9.scheduler配置信息

# 調度程序嘗試觸發新任務的時間
scheduler_heartbeat_sec = 60
# 檢測新dag的時間
min_file_process_interval = 10
# 是否使用catchup功能, 即是否執行自上次execute_date以來所有未執行的DAG Run, 另外定義每個DAG對象可傳遞catchup參數進行覆蓋
catchup_by_default = True

運行airflow

啟動schedule
airflow scheduler -D
啟動webserver
airflow webserver -D

安裝問題匯總

1. Global variable explicit_defaults_for_timestamp needs to be on (1) for mysql


修改Mysql配置文件my.cnf，具體步驟如下：

查找my.cnf文件位置
mysql --help | grep my.cnf
下圖紅框處為my.cnf文件所在位置：


修改文件
explicit_defaults_for_timestamp=true
注意：必須寫在【mysqld】下

重啟Mysql
sudo systemctl restart mysqld.service
查看修改是否生效。執行如下SQL，如果值為1則為生效。


2. pip install "apache-airflow[mysql]"報錯：

mysql_config not found

安裝mysql-devel:

首先查看是否有mysql_config文件。
find / -name mysql_config

如果沒有安裝mysql-devel
yum install mysql-devel
安裝之后再次查找，結果如圖：

3. 其他問題找我

配置任務

在 AirFlow 中，每個節點都是一個任務，可以是一條命令行 (BashOperator)，可以是一段 Python 腳本 (PythonOperator) 等等，然后這些節點根據依賴關系構成了一條流程，一個圖，稱為一個 DAG 。

默認會到 ${AIRFLOW_HOME}/dags 目錄下查找，可以直接在該目錄下創建相應的文件。

如下是一個簡單的示例。

import airflow
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
from datetime import timedelta, datetime
import pytz

# -------------------------------------------------------------------------------
# these args will get passed on to each operator
# you can override them on a per-task basis during operator initialization

default_args = {
    'owner': 'qxy',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

tz = pytz.timezone('Asia/Shanghai')
# naive = datetime.strptime("2018-06-13 17:40:00", "%Y-%m-%d %H:%M:%S")
# local_dt = tz.localize(naive, is_dst=None)
# utc_dt = local_dt.astimezone(pytz.utc).replace(tzinfo=None)

dt = datetime(2019, 7, 16, 16, 30, tzinfo=tz)
utc_dt = dt.astimezone(pytz.utc).replace(tzinfo=None)


dag = DAG(
    'airflow_interval_test',
    default_args=default_args,
    description='airflow_interval_test',
    schedule_interval='35 17 * * *',
    start_date=utc_dt

)

t1 = BashOperator(
    task_id='sleep',
    bash_command='sleep 5',
    dag=dag)

t2 = BashOperator(
    task_id='print_date',
    bash_command='date',
    dag=dag)

t1 >> t2

該文件創建一個簡單的 DAG，只有三個運算符，兩個 BaseOperator ，也就是執行 Bash 命令分別打印日期以及休眠 5 秒；另一個為 PythonOperator 在執行任務時調用 print_hello() 函數。
文件創建好后，放置到 ${AIRFLOW_HOME}/dags，airflow 自動讀取該DAG。

----- 測試是否正常，如果無報錯那么就說明正常
$ python /tmp/project/dags/hello_world.py