Pycharm调试Airflow

本文介绍如何使用Pycharm对Airflow进行Debug。
首先准备一个项目，目录如下：

├── airflow-project
│   ├── dags
│   │   └── debug_demo.py

其中debug_demo.py，包含如下内容，作为我们的调试对象

"""
## 演示debug
"""
from datetime import datetime, timedelta
from airflow.operators.python_operator import PythonOperator
from airflow.operators.dummy_operator import DummyOperator

from airflow import DAG

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2018, 10, 8),
}
dag = DAG('debug_demo', catchup=False, default_args=default_args, schedule_interval="*/5 * * * *")
dag.doc_md = __doc__


def foo(**context):
    print(len(context))
    raise Exception("故意抛出异常")


t1 = PythonOperator(python_callable=foo, task_id="foo", dag=dag)
t2 = DummyOperator(task_id="dummy", dag=dag)
t1.doc_md = """
### 一个task
"""
t2.doc_md = """
### 另一个task
"""
t1 >> t2

通过上面的程序，我们创建了一个dag_id为debug_demo的工作流，其中有两个顺序执行的节点，task_id依次是foo、dummy。其中taskfoo对应了python方法def foo(**context)，如果方法foo由复杂的代码组成，我们需要在其中打断点进行单步调试。
参照下图中的步骤，添加一个Run /Debug Configurations，在script path中选择airflow命令文件，Parameters中填写test {dag_id} {task_id} {execute_date}，这需要注意的是{execute_date}一定要精确到毫秒，否则会运行失败。

Pycharm配置

参考资料：

1. http://michal.karzynski.pl/blog/2017/03/19/developing-workflows-with-apache-airflow/