How to create a conditional task in Airflow

by

Airflow has a BranchPythonOperator that can be used to express the branching dependency more directly.

The docs describe its use:

The BranchPythonOperator is much like the PythonOperator except that it expects a python_callable that returns a task_id. The task_id returned is followed, and all of the other paths are skipped. The task_id returned by the Python function has to be referencing a task directly downstream from the BranchPythonOperator task.

…

If you want to skip some tasks, keep in mind that you can’t have an empty path, if so make a dummy task.

Code Example

def dummy_test():
    return 'branch_a'

A_task = DummyOperator(task_id='branch_a', dag=dag)
B_task = DummyOperator(task_id='branch_false', dag=dag)

branch_task = BranchPythonOperator(
    task_id='branching',
    python_callable=dummy_test,
    dag=dag,
)

branch_task >> A_task 
branch_task >> B_task

EDIT:

If you’re installing an Airflow version >=1.10.3, you can also return a list of task ids, allowing you to skip multiple downstream paths in a single Operator and don’t use a dummy task before joining.

More Related Contents:

  1. Conditional Replace Pandas
  2. Replace all elements of Python NumPy Array that are greater than some value
  3. How to do/workaround a conditional join in python Pandas?
  4. Proper way to create dynamic workflows in Airflow
  5. Conditional with statement in Python
  6. setting up s3 for logs in airflow
  7. How to have multiple conditions for one if statement in python [duplicate]
  8. Detecting Vowels vs Consonants In Python [duplicate]
  9. How do I assign values based on multiple conditions for existing columns?
  10. Set value of one Pandas column based on value in another column
  11. Multiple ‘or’ condition in Python [duplicate]
  12. How to do a conditional decorator in python?
  13. How to run Airflow on Windows
  14. “or” conditional in Python troubles [duplicate]
  15. pass task instance variables as op_args
  16. How to prevent airflow from backfilling dag runs?
  17. How to control the parallelism or concurrency of an Airflow installation?
  18. Airflow “This DAG isnt available in the webserver DagBag object “
  19. Assignment Condition in Python While Loop
  20. raise statement on a conditional expression
  21. Python pandas – new column’s value if the item is in the list
  22. Apache Airflow scheduler does not trigger DAG at schedule time
  23. Pandas: Join dataframe with condition
  24. Why is this simple conditional expression not working? [duplicate]
  25. Is there a NumPy function to return the first index of something in an array?
  26. random.seed(): What does it do?
  27. Python: How to match nested parentheses with regex?
  28. Evenly distributing n points on a sphere
  29. Item frequency count in Python
  30. Python file.write creating extra carriage return

Leave a Comment