The Airflow BranchPythonOperator for Beginners in 10 mins - Execute specific tasks to execute.
👍 Smash the like button to become better at Airflow
❤️ Subscribe to my channel to become a master of Airflow
🏆 Take my course : www.udemy.com/course/the-ulti... to join the legends of Airflow
🚨 My Patreon: / marclamberti to support my work and be friend for life
The materials: www.notion.so/The-BranchPytho...
The blogs post: marclamberti.com/blog/airflow...
1. Use case
You have a data pipelines with 3 different tasks. The first is_api_available checks if the API is available or not. Next, if the API is available you execute the task Download, but if NOT you want to execute is_api_available_2 first and then the task Download. How can you do that? How can you choose what task or another according to a condition? The BranchPythonOperator!
2. What is the BranchPythonOperator?
The BranchPythonOperator allows to choose a specific path in your DAG according to a criterion. More specifically, it returns the task id of the NEXT task to execute based on a condition/value etc.
3. How to use the BranchPythonOperator
Really simple, you import the BranchPythonOperator. You implement it.
You have to pass a python callable function. That function is triggered by the BranchPythonOperator like with the PythonOperator but this time, the function must return the task id of the next task to execute. Finally, specify the dependencies between your tasks.
4. The BranchPythonOperator with XComs
Be careful. Each time a BranchPythonOperator is triggered, 2 XComs are created. XComs are not removed automatically. If you have hundreds of BranchPythonOperators, you will store a lot of XComs. To mitigate this, specify the parameter do_xcom_push in the BranchPythonOperator to only generate one XCom instead of one.
5. BranchPythonOperator task skipped
When a task is not triggered by the BranchPythonOperator, it is skipped. The problem is, if you have a task which depends both on the skipped task and the task triggered by the BranchPythonOperator, that task will be skipped as well. Why? Trigger rules! By default, if one of the parents of a task is skipped, then the task is skipped as well. You can fix this with the trigger rule none_failed_or_skipped
6. BranchPythonOperator multiple task ids
You can definitely choose to execute multiple tasks with the BranchPythonOperator. You just have to return a list of task ids instead of a task id.
Enjoy!