如何强制气流任务失败？

Question

我有一个python可调process_csv_entries用来处理csv文件条目.只有成功处理完所有条目后,我才希望我的任务成功完成.否则任务应该失败

def process_csv_entries(csv_file):
    # Boolean 
    file_completely_parsed = <call_to_module_to_parse_csv>
    return not file_completely_parsed

CSV_FILE=<Sets path to csv file>
t1 = PythonOperator(dag=dag,
                      task_id='parse_csv_completely',
                      python_operator=process_csv_entries,
                      op_args=[CSV_FILE])

无论返回值如何,t1似乎都成功完成.如何强制PythonOperator任务失败？

Answer 1

如果您想让任务失败而不重试，请使用AirflowFailException：-

例子 ：-

from airflow.exceptions import AirflowFailException
def task_to_fail():
    raise AirflowFailException("Our api key is bad!")

如果您正在寻找重试，请使用AirflowException：-

例子：-

from airflow import AirflowException
def task_to_fail():
    raise AirflowException("Error msj")

Answer 2

遇到错误条件时引发异常(在您的情况下:文件未被解析时)

raise ValueError('File not parsed completely/correctly')

用适当的消息提出相关的错误类型

Answer 3

是的,加注AirflowException,这将导致任务立即移动到失败状态.

from airflow import AirflowException

ValueError 可用于失败和重试.

Answer 4

AirflowFailException现在可用于使任务失败而无需重试