本机Airflow 安装
# 1. 配置airflow环境变量
echo '\nexport AIRFLOW_HOME=~/opt/airflow' > ~/.bashrc
source ~/.bashrc
# 2. 安装airflow
pip install apache-airflow
airflow initdb
# 3. 启动airflow
airflow webserver -D
airflow scheduler -D
# 4. 创建dags目录
mkdir ~/opt/airflow/dags
Pycharm 配置
设置Interpreter
python env 选择安装airflow的python， 这样pycharm的环境才会包含airflow的依赖。
image-20201110140343519.png
image-20201110140410115.png
新建Dag
pycharm打开 ${AIRFLOW_HOME} 目录，在dags目录下新建hello dag
image-20201110141137000.png
# hello.py
# -*- coding: utf-8 -*-
from datetime import timedelta
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.utils.dates import days_ago
args = {
'owner': 'airflow',
}
dag = DAG(
dag_id='hello',
default_args=args,
schedule_interval=None,
start_date=days_ago(1),
dagrun_timeout=timedelta(minutes=60)
)
def print_hello():
print('Hello airflow!!')
return 'Hello airflow!!'
dummy_operator = DummyOperator(task_id='start', retries=3, dag=dag)
hello_operator = PythonOperator(task_id='hello', python_callable=print_hello, dag=dag)
dummy_operator >> hello_operator
Airflow test
# airflow test
# usage: airflow test [-h] [-sd SUBDIR] [-dr] [-tp TASK_PARAMS] [-pm] dag_id task_id execution_date
airflow test hello hello 2020-10-10
image-20201110141334555.png
Pycharm run
image-20201110142744922.png
然后run hello.py
image-20201110142905241.png
