计划的 DAG 未运行。我如何诊断问题？

Question

我使用的是 Airflow 1.8.1。我有一个 DAG，我相信我已安排每 5 分钟运行一次，但它没有这样做：

忽略两次成功的 DAG 运行，它们是手动触发的。

我查看了该 DAG 的调度程序日志，我看到了：

[2019-04-26 22:03:35,601] {jobs.py:343} DagFileProcessor839 INFO - Started process (PID=5653) to work on /usr/local/airflow/dags/retrieve_airflow_artifacts.py
[2019-04-26 22:03:35,606] {jobs.py:1525} DagFileProcessor839 INFO - Processing file /usr/local/airflow/dags/retrieve_airflow_artifacts.py for tasks to queue
[2019-04-26 22:03:35,607] {models.py:168} DagFileProcessor839 INFO - Filling up the DagBag from /usr/local/airflow/dags/retrieve_airflow_artifacts.py
[2019-04-26 22:03:36,083] {jobs.py:1539} DagFileProcessor839 INFO - DAG(s) ['retrieve_airflow_artifacts'] retrieved from /usr/local/airflow/dags/retrieve_airflow_artifacts.py
[2019-04-26 22:03:36,112] {jobs.py:1172} DagFileProcessor839 INFO - Processing retrieve_airflow_artifacts
[2019-04-26 22:03:36,126] {jobs.py:566} DagFileProcessor839 INFO - Skipping SLA check for <DAG: retrieve_airflow_artifacts> because no tasks in DAG have SLAs
[2019-04-26 22:03:36,132] {models.py:323} DagFileProcessor839 INFO - Finding 'running' jobs without a recent heartbeat
[2019-04-26 22:03:36,132] {models.py:329} DagFileProcessor839 INFO - Failing jobs without heartbeat after 2019-04-26 21:58:36.132768
[2019-04-26 22:03:36,139] {jobs.py:351} DagFileProcessor839 INFO - Processing /usr/local/airflow/dags/retrieve_airflow_artifacts.py took 0.539 seconds
[2019-04-26 22:04:06,776] {jobs.py:343} DagFileProcessor845 INFO - Started process (PID=5678) to work on /usr/local/airflow/dags/retrieve_airflow_artifacts.py
[2019-04-26 22:04:06,780] {jobs.py:1525} DagFileProcessor845 INFO - Processing file /usr/local/airflow/dags/retrieve_airflow_artifacts.py for tasks to queue
[2019-04-26 22:04:06,780] {models.py:168} DagFileProcessor845 INFO - Filling up the DagBag from /usr/local/airflow/dags/retrieve_airflow_artifacts.py
[2019-04-26 22:04:07,258] {jobs.py:1539} DagFileProcessor845 INFO - DAG(s) ['retrieve_airflow_artifacts'] retrieved from /usr/local/airflow/dags/retrieve_airflow_artifacts.py
[2019-04-26 22:04:07,287] {jobs.py:1172} DagFileProcessor845 INFO - Processing retrieve_airflow_artifacts
[2019-04-26 22:04:07,301] {jobs.py:566} DagFileProcessor845 INFO - Skipping SLA check for <DAG: retrieve_airflow_artifacts> because no tasks in DAG have SLAs
[2019-04-26 22:04:07,307] {models.py:323} DagFileProcessor845 INFO - Finding 'running' jobs without a recent heartbeat
[2019-04-26 22:04:07,307] {models.py:329} DagFileProcessor845 INFO - Failing jobs without heartbeat after 2019-04-26 21:59:07.307607
[2019-04-26 22:04:07,314] {jobs.py:351} DagFileProcessor845 INFO - Processing /usr/local/airflow/dags/retrieve_airflow_artifacts.py took 0.538 seconds

一遍又一遍。我已经将它与另一台服务器上的 DAG 进行了比较，通过这样做我知道会有额外的日志记录表明 DAG 已通过计划触发，但此日志文件中没有此类记录。

以下是我的 DAG 计划的定义方式：

args = {
    'owner': 'airflow',
    'start_date': (datetime.datetime.now() - datetime.timedelta(minutes=5))
}

dag = DAG(
    dag_id='retrieve_airflow_artifacts', default_args=args,
    schedule_interval="0,5,10,15,20,25,30,35,40,45,50,55 * * * *")

有人能帮我弄清楚为什么我的 DAG 没有运行吗，因为我看了很多次都无法弄清楚。

Answer 1

如果我不得不猜测，我会说您的 start_date 给您带来了一些问题。

将您的参数更改为静态启动并防止它在过去的时间间隔内运行：

args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2019, 4, 27) #year month day
}

另外，为了更容易阅读，将你的 DAG 参数更改为（相同的功能）：

dag = DAG(
    dag_id='retrieve_airflow_artifacts', 
    default_args=args,
    schedule_interval="*/5 * * * *"
)

这应该允许调度程序来接它！

一般建议不要动态设置 start_date。

取自Airflow FAQ：

我们建议不要使用动态值作为 start_date，尤其是 datetime.now() 因为它可能会很混乱。任务被触发 一旦周期结束，理论上@hourly DAG 永远不会得到 随着 now() 的移动，到现在一小时后。

关于此的另一个 SO 问题：why dynamic start dates cause issues