60

I have a python callable process_csv_entries that processes csv file entries. I want my task to complete successfully only if all entries were processed successfully. Task should fail otherwise

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 seems to complete successfully irrespective of returned value. How do I force PythonOperator task to fail?

4 Answers 4

55

raise exception when you meet the error condition ( in your case: when file is not sucesfully parsed)

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

raise relevant error type with suitable message

2
  • 3
    This works..Thanks! I was hoping there would be a better way to handle this.
    – Mask
    Commented Mar 30, 2017 at 20:38
  • 16
    This will not force the task instance status to "failed" though... Is there a way to bypass the retries config?
    – c-a
    Commented May 25, 2018 at 16:41
52

Yes, raise AirflowException, this will cause the task to move immediately to failure state.

from airflow import AirflowException

ValueError can be used for fail and retry.

4
  • 2
    Any exception fails the task and moves it to failure state.
    – bcb
    Commented Dec 14, 2018 at 0:59
  • 18
    Is there a way to prevent airflow from retrying the task if retries are set? For example, there are some errors that you don't want/need to retry, such as invalid input related errors. Commented Apr 9, 2019 at 17:08
  • 2
    @JoeJ, there is a PR open for that: github.com/apache/airflow/pull/7133
    – gabra
    Commented Mar 16, 2020 at 17:35
  • AirflowException also avoids printing the stack trace on the logs +1 Commented Jul 13, 2023 at 9:01
34

if you want to fail the task without retries use AirflowFailException :-

Example :-

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

If you are looking for retries use AirflowException :-

Example:-

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

AirflowFailException is now available in Airflow 1.10.11 to make the task fail without retries

1
  • 7
    For those of you wondering what version "now" means, it was introduced in 1.10.11
    – ZaxR
    Commented Jun 25, 2021 at 15:30

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.