本机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