Why branching handles conditional logic in Apache Airflow - Test Your Understanding

Practice - 5 Tasks

1fill in blank

easy

Complete the code to create a BranchPythonOperator that decides the next task based on a condition.

Apache Airflow

branch_task = BranchPythonOperator(task_id='branching', python_callable=[1], dag=dag)

Drag options to blanks, or click blank then click option'

Abranch_logic

Brun_task

Cchoose_branch

Dstart_branch

Attempts:

3 left

💡 Hint

Common Mistakes

Using a string instead of a function name

Passing a task instead of a callable

The python_callable parameter must be a function that returns the task_id to follow. Here, choose_branch is the function that handles the condition.

2fill in blank

medium

Complete the function to return the correct task_id based on the value of condition.

Apache Airflow

def choose_branch():
    condition = True
    if condition:
        return [1]
    else:
        return 'task_b'

Drag options to blanks, or click blank then click option'

A'task_a'

B'task_c'

C'branch_task'

D'start_task'

Attempts:

3 left

💡 Hint

Common Mistakes

Returning a variable instead of a string

Missing quotes around the task_id

The function must return the task_id string of the next task to run. If condition is True, it returns 'task_a'.

3fill in blank

hard

Fix the error in the BranchPythonOperator definition by completing the missing parameter.

Apache Airflow

branch_task = BranchPythonOperator(task_id='branching', python_callable=choose_branch, [1]=dag)

Drag options to blanks, or click blank then click option'

Astart_date

Bschedule_interval

Cdefault_args

Ddag

Attempts:

3 left

💡 Hint

Common Mistakes

Using schedule_interval instead of dag

Omitting the DAG parameter

The dag parameter links the operator to the DAG object. Without it, the task is not part of the DAG.

4fill in blank

hard

Fill both blanks to create a DAG with a BranchPythonOperator and two downstream tasks.

Apache Airflow

with DAG('branching_dag', start_date=days_ago(1), schedule_interval='@daily') as dag:
    branch = BranchPythonOperator(task_id='branch', python_callable=[1])
    task_a = DummyOperator(task_id=[2])
    task_b = DummyOperator(task_id='task_b')
    branch >> [task_a, task_b]

Drag options to blanks, or click blank then click option'

Achoose_branch

B'task_a'

C'branch_task'

Drun_branch

Attempts:

3 left

💡 Hint

Common Mistakes

Mismatching task_id and return value

Wrong function name in callable

The python_callable must be the function choose_branch. The task_a DummyOperator must have the task_id 'task_a' to match the branch return value.

5fill in blank

hard

Fill all three blanks to define a branching function that returns different task_ids based on a variable.

Apache Airflow

def choose_branch():
    value = 10
    if value > [1]:
        return [2]
    else:
        return [3]

Drag options to blanks, or click blank then click option'

A5

B'task_high'

C'task_low'

D15

Attempts:

3 left

💡 Hint

Common Mistakes

Using numbers instead of strings for task_ids

Wrong comparison operator

The function checks if value is greater than 5. If true, it returns 'task_high', else 'task_low'.

Practice

1. What is the main purpose of branching in Airflow workflows?

easy

A. To choose which tasks run based on conditions

B. To run all tasks in parallel

C. To schedule tasks at fixed intervals

D. To retry failed tasks automatically

Solution

1. Step 1: Understand branching concept

   Branching allows workflows to decide which path to take based on conditions.

2. Step 2: Compare with other options

   Running all tasks in parallel or scheduling are different features, not branching.

3. Final Answer:

   To choose which tasks run based on conditions -> Option A

4. Quick Check:

   Branching = Conditional path selection [OK]

Hint: Branching picks one path based on condition [OK]

Common Mistakes:

• Confusing branching with parallel execution
• Thinking branching schedules tasks
• Assuming branching retries tasks

2. Which Airflow operator is used to implement branching logic?

easy

A. PythonOperator

B. DummyOperator

C. BashOperator

D. BranchPythonOperator

Solution

1. Step 1: Identify branching operator

   The BranchPythonOperator is designed to choose the next task based on a Python function's return value.

2. Step 2: Eliminate other operators

   PythonOperator runs Python code but does not branch; DummyOperator is a placeholder; BashOperator runs shell commands.

3. Final Answer:

   BranchPythonOperator -> Option D

4. Quick Check:

   Branching operator = BranchPythonOperator [OK]

Hint: BranchPythonOperator controls conditional task flow [OK]

Common Mistakes:

• Choosing PythonOperator instead of BranchPythonOperator
• Confusing DummyOperator as branching
• Using BashOperator for branching

3. Given this BranchPythonOperator function, which task will run next?

def choose_task():
    if 5 > 3:
        return 'task_a'
    else:
        return 'task_b'

medium

A. task_b

B. Both task_a and task_b

C. task_a

D. No task runs

Solution

1. Step 1: Evaluate the condition in the function

   5 > 3 is True, so the function returns 'task_a'.

2. Step 2: Understand branching behavior

   Only the task returned by the function runs; others are skipped.

3. Final Answer:

   task_a -> Option C

4. Quick Check:

   Condition true -> task_a runs [OK]

Hint: Return task name to run next [OK]

Common Mistakes:

• Thinking both tasks run
• Choosing task_b despite condition
• Assuming no task runs

4. What is wrong with this BranchPythonOperator function?

def choose_task():
    if x > 10:
        return 'task_x'
    else:
        return 'task_y'

medium

A. Variable x is not defined, causing an error

B. The function does not return a task id

C. BranchPythonOperator cannot return strings

D. The else block is missing

Solution

1. Step 1: Check variable usage

   The variable x is used but not defined anywhere in the function or globally.

2. Step 2: Understand error cause

   Using an undefined variable causes a NameError at runtime.

3. Final Answer:

   Variable x is not defined, causing an error -> Option A

4. Quick Check:

   Undefined variable = runtime error [OK]

Hint: Check all variables are defined before use [OK]

Common Mistakes:

• Assuming missing else block causes error
• Thinking BranchPythonOperator can't return strings
• Ignoring undefined variable errors

5. You want to run task_a if a file exists, otherwise run task_b. Which BranchPythonOperator function correctly implements this?

hard

A. def choose():
if os.path.exists('/path/file'):
print('task_a')
else:
print('task_b')

B. def choose():
if os.path.exists('/path/file'):
return 'task_a'
return 'task_b'

C. def choose():
if os.path.exists('/path/file'):
return ['task_a', 'task_b']
else:
return 'task_b'

D. def choose():
if os.path.exists('/path/file'):
return 'task_b'
else:
return 'task_a'

Solution

1. Step 1: Check condition and return values

   def choose(): if os.path.exists('/path/file'): return 'task_a' return 'task_b' returns 'task_a' if file exists, else 'task_b', matching the requirement.

2. Step 2: Identify errors in other options

   def choose(): if os.path.exists('/path/file'): return 'task_b' else: return 'task_a' reverses tasks; def choose(): if os.path.exists('/path/file'): return ['task_a', 'task_b'] else: return 'task_b' returns a list which is invalid; def choose(): if os.path.exists('/path/file'): print('task_a') else: print('task_b') prints instead of returning task id.

3. Final Answer:

   def choose(): if os.path.exists('/path/file'): return 'task_a' return 'task_b' -> Option B

4. Quick Check:

   Return correct task id based on condition [OK]

Hint: Return one task id string, not list or print [OK]

Common Mistakes:

• Returning list of tasks instead of single task
• Printing task names instead of returning
• Swapping task names in condition