时间戳

首页 > 解决方案 > AirFlow DAG 卡在运行状态

airflow - AirFlow DAG 卡在运行状态

问题描述

我创建了一个 dag 并每天安排它。它每天都在排队，但任务实际上并没有运行。过去这里已经提出了这个问题，但答案对我没有帮助，所以似乎还有另一个问题。

我的代码在下面共享。我用注释替换了任务 t2 的 SQL。当我使用“气流测试...”在 CLI 上单独运行每个任务时，每个任务都会成功运行。

你能解释一下应该做些什么来让 DAG 运行吗？谢谢！

这是 DAG 代码：

from datetime import timedelta, datetime
from airflow import DAG
from airflow.contrib.operators.bigquery_operator import BigQueryOperator



default_args = {
    'owner' : 'me',
    'depends_on_past' : 'true',
    'start_date' : datetime(2018, 06, 25),
    'email' : ['myemail@moovit.com'],
    'email_on_failure':True,
    'email_on_retry':False,
    'retries' : 2,
    'retry_delay' : timedelta(minutes=5)
}


dag = DAG('my_agg_table',
default_args = default_args,
schedule_interval = "30 4 * * *"
)



t1 = BigQueryOperator(
    task_id='bq_delete_my_agg_table',
    use_legacy_sql=False,
    write_disposition='WRITE_TRUNCATE',
    allow_large_results=True,
    bql='''
    delete `my_project.agg.my_agg_table`
    where date = '{{ macros.ds_add(ds, -1)}}'
    ''',
    dag=dag)

t2 = BigQueryOperator(
    task_id='bq_insert_my_agg_table',
    use_legacy_sql=False,
    write_disposition='WRITE_APPEND',
    allow_large_results=True,
    bql='''
    #standardSQL
    Select ... the query continue here.....
    ''',    destination_dataset_table='my_project.agg.my_agg_table',
    dag=dag)


t1 >> t2

标签： airflow

解决方案

通常很容易找出任务未运行的原因。在 Airflow 网页界面中时：

  • 选择任何感兴趣的 DAG
  • 现在点击任务
  • 再次点击Task Instance Details
  • 在第一行有一个面板Task Instance State
  • 在旁边的框中Reason是运行任务的原因 - 或忽略任务的原因

检查第一个没有执行的任务通常是有意义的，因为我看到你有设置depends_on_past=True，如果在错误的场景中使用可能会导致问题。

更多信息在这里：Airflow 1.9.0 正在排队但不启动任务

推荐阅读