PythonOperator for custom logic in Apache Airflow - Practice Problems & Coding Challenges

Challenge - 5 Problems

💻 Command Output

intermediate

Output of PythonOperator task execution

Given this Airflow DAG snippet, what will be the output logged when the python_task runs?

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def greet():
    print('Hello from Airflow!')

default_args = {'start_date': datetime(2024, 1, 1)}

dag = DAG('greet_dag', default_args=default_args, schedule_interval='@daily')

python_task = PythonOperator(
    task_id='greet_task',
    python_callable=greet,
    dag=dag
)

A Task greet_task failed due to missing return

B greet_task executed successfully

C No output is logged by PythonOperator

D Hello from Airflow!

💡 Hint

Think about what the print statement inside the python_callable does.

The PythonOperator runs the greet function, which prints 'Hello from Airflow!'. This output appears in the task logs.

❓ Configuration

intermediate

Correct PythonOperator usage with arguments

Which option correctly passes arguments to a Python function using PythonOperator?

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def add_numbers(a, b):
    print(a + b)

default_args = {'start_date': datetime(2024, 1, 1)}

dag = DAG('add_dag', default_args=default_args, schedule_interval='@daily')

A PythonOperator(task_id='add_task', python_callable=add_numbers, op_kwargs={'a': 5, 'b': 3}, dag=dag)

B PythonOperator(task_id='add_task', python_callable=add_numbers(5, 3), dag=dag)

C PythonOperator(task_id='add_task', python_callable=add_numbers, args=(5, 3), dag=dag)

D PythonOperator(task_id='add_task', python_callable=add_numbers, op_args=[5, 3], dag=dag)

💡 Hint

Check how keyword arguments are passed to the callable in Airflow.

op_kwargs is the correct way to pass named arguments to the Python function in PythonOperator. Options A and C use positional arguments but only op_args is valid for positional, and option A calls the function immediately which is incorrect.

❓ Troubleshoot

advanced

Error raised by PythonOperator with missing callable

What error will Airflow raise if python_callable is set to None in PythonOperator?

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

default_args = {'start_date': datetime(2024, 1, 1)}

dag = DAG('error_dag', default_args=default_args, schedule_interval='@daily')

python_task = PythonOperator(task_id='error_task', python_callable=None, dag=dag)

A AirflowSkipException: Task skipped due to None callable

B ValueError: python_callable must be a callable

C TypeError: 'NoneType' object is not callable

D No error, task runs with no operation

💡 Hint

Think about what happens when Python tries to call None as a function.

Setting python_callable=None means Airflow tries to call None as a function, which raises a TypeError because NoneType is not callable.

🔀 Workflow

advanced

Order of execution with PythonOperator tasks

Given these two PythonOperator tasks in a DAG, which option shows the correct order of execution?

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def task1():
    print('Task 1')

def task2():
    print('Task 2')

default_args = {'start_date': datetime(2024, 1, 1)}

dag = DAG('order_dag', default_args=default_args, schedule_interval='@daily')

t1 = PythonOperator(task_id='task1', python_callable=task1, dag=dag)
t2 = PythonOperator(task_id='task2', python_callable=task2, dag=dag)

t1 >> t2

A Task 1 runs before Task 2

B Task 2 runs before Task 1

C Tasks run in parallel with no order

D Task 2 runs twice, Task 1 once

💡 Hint

Look at the dependency operator >> between tasks.

The operator >> means Task 1 must finish before Task 2 starts, so Task 1 runs first, then Task 2.

✅ Best Practice

expert

Best practice for returning values from PythonOperator

Which option correctly captures the return value of a Python function executed by PythonOperator for use in downstream tasks?

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.python import get_current_context
from datetime import datetime

def generate_value():
    return 42

def print_value():
    context = get_current_context()
    value = context['ti'].xcom_pull(task_ids='generate_task')
    print(f'Received value: {value}')

default_args = {'start_date': datetime(2024, 1, 1)}

dag = DAG('xcom_dag', default_args=default_args, schedule_interval='@daily')

generate_task = PythonOperator(task_id='generate_task', python_callable=generate_value, dag=dag)
print_task = PythonOperator(task_id='print_task', python_callable=print_value, dag=dag)
generate_task >> print_task

A Store value in a global variable accessible by both tasks

B Use return statement in the first task and pull with xcom_pull in the second task

C Write value to a file in the first task and read it in the second task

D Pass value as an argument directly to the second task's python_callable

💡 Hint

Think about how Airflow shares data between tasks.

Airflow uses XComs to pass data between tasks. Returning a value from a PythonOperator stores it in XCom automatically. The downstream task can retrieve it using xcom_pull.

Practice

1. What is the main purpose of using PythonOperator in Airflow?

easy

A. To run custom Python functions as tasks in a workflow

B. To schedule workflows using cron syntax

C. To monitor system resources during task execution

D. To create database connections automatically

Solution

1. Step 1: Understand PythonOperator's role

   PythonOperator allows you to run your own Python code as a task inside Airflow workflows.

2. Step 2: Compare with other options

   Scheduling uses other operators or settings, monitoring and DB connections are separate features.

3. Final Answer:

   To run custom Python functions as tasks in a workflow -> Option A

4. Quick Check:

   PythonOperator runs Python code = B [OK]

Hint: PythonOperator runs your Python function inside Airflow [OK]

Common Mistakes:

• Confusing PythonOperator with scheduling features
• Thinking it manages database connections
• Assuming it monitors system resources

2. Which of the following is the correct way to pass arguments to a Python function using PythonOperator?

easy

A. PythonOperator(task_id='task', python_callable=my_func, parameters=[1, 2])

B. PythonOperator(task_id='task', python_callable=my_func, args=[1, 2])

C. PythonOperator(task_id='task', python_callable=my_func, op_args=[1, 2])

D. PythonOperator(task_id='task', python_callable=my_func, inputs=[1, 2])

Solution

1. Step 1: Recall argument passing syntax

   PythonOperator uses op_args to pass positional arguments to the function.

2. Step 2: Check options for correctness

   Only PythonOperator(task_id='task', python_callable=my_func, op_args=[1, 2]) uses op_args correctly; others use wrong parameter names.

3. Final Answer:

   PythonOperator(task_id='task', python_callable=my_func, op_args=[1, 2]) -> Option C

4. Quick Check:

   Use op_args to pass args [OK]

Hint: Use op_args to pass positional arguments to your function [OK]

Common Mistakes:

• Using args instead of op_args
• Using parameters or inputs which are invalid
• Confusing op_args with op_kwargs

3. Given this code snippet, what will be printed when the task runs?

def greet(name):
    print(f"Hello, {name}!")

PythonOperator(
    task_id='say_hello',
    python_callable=greet,
    op_args=['Alice'],
    dag=dag
)

medium

A. Hello, ['Alice']!

B. Hello, name!

C. No output, function not called

D. Hello, Alice!

Solution

1. Step 1: Understand function call with op_args

   The function greet is called with 'Alice' as the argument for name.

2. Step 2: Evaluate the print statement output

   It prints "Hello, Alice!" exactly as formatted by the f-string.

3. Final Answer:

   Hello, Alice! -> Option D

4. Quick Check:

   Function prints Hello, Alice! [OK]

Hint: op_args passes list items as function arguments [OK]

Common Mistakes:

• Thinking op_args passes the list as one argument
• Confusing variable name with string 'name'
• Assuming function is not called

4. Identify the error in this PythonOperator usage:

def add(x, y):
    return x + y

PythonOperator(
    task_id='add_task',
    python_callable=add,
    op_kwargs={'x': 5, 'z': 3},
    dag=dag
)

medium

A. python_callable must be a string

B. The function add does not accept argument 'z'

C. task_id is missing

D. op_kwargs should be a list, not a dict

Solution

1. Step 1: Check function parameters

   The function add expects only 'x' and 'y' as arguments.

2. Step 2: Analyze op_kwargs keys

   op_kwargs passes 'x' and 'z', but 'z' is not a valid parameter for add, causing an error.

3. Final Answer:

   The function add does not accept argument 'z' -> Option B

4. Quick Check:

   Extra argument 'z' causes error [OK]

Hint: Function args must match op_kwargs keys exactly [OK]

Common Mistakes:

• Passing wrong argument names in op_kwargs
• Thinking op_kwargs can be a list
• Confusing python_callable type

5. You want to create a PythonOperator task that calls a function process_data which requires a list of numbers and a flag. The function signature is def process_data(numbers, verbose=False):. How do you correctly pass the arguments to the operator?

hard

A. PythonOperator(task_id='proc', python_callable=process_data, op_kwargs={'numbers': [1,2,3], 'verbose': True}, dag=dag)

B. PythonOperator(task_id='proc', python_callable=process_data, op_args=[1,2,3, True], dag=dag)

C. PythonOperator(task_id='proc', python_callable=process_data, op_args=[1,2,3], op_kwargs={'verbose': True}, dag=dag)

D. PythonOperator(task_id='proc', python_callable=process_data, args=[[1,2,3], True], dag=dag)

Solution

1. Step 1: Understand function parameters

   process_data expects 'numbers' (list) and optional 'verbose' keyword argument.

2. Step 2: Match argument passing style

   Using op_kwargs with keys 'numbers' and 'verbose' correctly passes both arguments by name.

3. Step 3: Check other options

   B passes four positional arguments causing TypeError. C passes three positional arguments (1,2,3) plus verbose=True, but function expects only two positional arguments causing TypeError. D uses invalid 'args' parameter.

4. Final Answer:

   PythonOperator(task_id='proc', python_callable=process_data, op_kwargs={'numbers': [1,2,3], 'verbose': True}, dag=dag) -> Option A

5. Quick Check:

   Use op_kwargs for named args [OK]

Hint: Use op_kwargs for named arguments, op_args for positional [OK]

Common Mistakes:

• Passing too many positional arguments
• Using wrong parameter names like args instead of op_args
• Mixing positional and keyword args incorrectly