Cloud Composer Airflow任务失败，但功能已成功完成

Question

我正试图用cloudcomposer编写我的第一个气流作业。我的DAG有三个任务，第一个任务成功完成，但第二个任务似乎失败了，并发出任何失败错误消息。我在第二个任务中使用PythonOperator。执行长查询，直到查询完成为止。查询完成后，我会收到一条消息，说明数据已输出到正确的表中，但Airflow将任务视为失败，并再次重试该任务。在

我的DAG的default_args如下所示：

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': today.strftime("%Y-%m-%d"),
    'email': ['email@email.com'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=1),
    'dagrun_timeout': timedelta(minutes=30)
}

编辑：

这是我的Python callable和PythonOperator。run_query可调用函数在Stackdriver日志中生成输出，并指示实际函数已完成，但任务失败。在

^{pr2}$

如果有任何提示，我将不胜感激！在

Answer 1

我在你的代码中看不到任何错误处理。在

当长时间运行的查询和轮询失败时，引发AirflowException，这将导致任务立即移到failure状态。在

from airflow import AirflowException

ValueError可用于失败和重试