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.