The Airflow BranchPythonOperator for Beginners in 10 mins

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!

[@MarcLamberti]

Comment below the next topic you would like to address :)

[@allwinjayson]

Please share the video for holidays excluding.
My use case is the job will be run on every 5th of month. If 5th is in my custom holiday list the job will run on the next working days... Please put the video which is very helpful for mw

[@cherrejim]

super well explained. Thanks

[@nicolaschouali4295]

Hi mark !! Very helpful. Thank you

[@matrix_AI]

The problem that u described with storing, I was facing that...... Thanks for helping

[@emre_0134]

It was very helpful , many thanks Marc🤩👏👌

[@MarcLamberti]

Thank you 🙏

[@sharmaakarsh]

Wonderful video 👌👌👌

[@rahulkumarpatnaik6023]

Thank you ❤️🙏

[@v4ldevrr4m47]

Thanks a lt for this material. Regarding next topic.I am curius about thitd libs in your PythonOperators scripts . considering tha my airflow is a container how it manage pip install ?

[@taukirkhan3417]

Very helpful

[@Arieleyo]

very useful! suscribed ❤

[@MarcLamberti]

❤️

[@siddhardhakommineni7390]

The input value accurate , Is there any way that I can pass it using UI input ??

[@mbcebrix]

Hi mark. Can you do indepth tutorial on creating dynamic dags?

[@MarcLamberti]

Look at my latest video 😉

[@user-sl9ot]

Do yo have an example about "6. BranchPythonOperator multiple task ids" ?

[@ahmedismail3828]

Hi Marc, I've been following your tutorials to get a grip on airflow, and I must say these are the best available but I keep getting a syntax error at python_callable line with the BranchPythonOperator, this way the dag doesn't get imported to airflow. Can you help out with this?

[@MarcLamberti]

Show me what you've done :)

[@mnsosa]

how tf do i use branching with dag decorator?

[@donnillorussia]

How to use branching between single task and a TaskGroup, especially if a TaskGroup is designed with TaskFlowAPI?!!!!!!!!!!! And where is task.branch decorator???????!!!!!