PythonOperator for custom logic in Apache Airflow - Step-by-Step Execution

Process Flow - PythonOperator for custom logic

Define Python function

↓

Create PythonOperator with function

↓

Add operator to DAG

↓

DAG runs -> PythonOperator triggers

↓

Function executes custom logic

↓

Task completes with result

This flow shows how a Python function is wrapped by PythonOperator, added to a DAG, and executed when the DAG runs.

Execution Sample

Apache Airflow

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

from airflow.operators.python import PythonOperator

from airflow import DAG
from datetime import datetime

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

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

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

Defines a simple function and creates a PythonOperator that runs it as a task.

Process Table

Step	Action	Evaluation	Result
1	Define function greet	Function greet() created	Function ready to use
2	Create PythonOperator greet_task	Assign greet as python_callable	Operator ready with task_id 'greet'
3	Add greet_task to DAG	DAG includes greet_task	Task scheduled in DAG
4	DAG run starts	PythonOperator triggers greet	Task execution begins
5	Execute greet()	Print statement runs	Output: 'Hello from Airflow!'
6	Task completes	No errors	Task marked success

💡 Task completes successfully after running the custom Python function.

Status Tracker

Variable	Start	After Step 1	After Step 2	After Step 3	After Step 4	After Step 5	Final
greet	undefined	function greet()	function greet()	function greet()	function greet()	function greet()	function greet()
greet_task	undefined	undefined	PythonOperator with greet	PythonOperator with greet	PythonOperator with greet	PythonOperator with greet	PythonOperator with greet
DAG	undefined	undefined	undefined	DAG with greet_task	DAG running greet_task	DAG running greet_task	DAG completed greet_task

Key Moments - 3 Insights

Why do we pass the function name without parentheses to python_callable?

Because python_callable expects a reference to the function itself, not the result of calling it. See execution_table step 2 where greet is assigned without () to keep it callable by Airflow.

What happens if the function prints output? Does Airflow capture it?

Yes, Airflow captures standard output in task logs. In execution_table step 5, the print runs and its output is stored in logs accessible via Airflow UI.

How does Airflow know when the task is done?

Airflow marks the task success after the function completes without error, as shown in execution_table step 6.

Visual Quiz - 3 Questions

Look at the execution table, what is the output at step 5?

A Error message

B No output

C 'Hello from Airflow!'

D 'Task completed'

💡 Hint

Check the 'Result' column at step 5 in the execution_table.

At which step does Airflow start running the Python function?

A Step 2

B Step 4

C Step 1

D Step 6

💡 Hint

Look for when the DAG run triggers the PythonOperator in the execution_table.

If you mistakenly write python_callable=greet() instead of python_callable=greet, what happens?

A Function runs immediately during DAG definition

B Task runs normally when triggered

C Airflow throws syntax error

D Task is skipped

💡 Hint

Think about when the function is called if parentheses are used, referencing key_moments about function references.

Concept Snapshot

PythonOperator runs custom Python code in Airflow.
Define a Python function.
Pass it to PythonOperator via python_callable (no parentheses).
Add operator to DAG.
When DAG runs, operator calls your function.
Output is logged; task success depends on function completion.

Full Transcript

This visual execution shows how to use PythonOperator in Airflow to run custom Python logic. First, you define a Python function with the code you want to run. Then, you create a PythonOperator and assign your function to its python_callable parameter without parentheses, so Airflow can call it later. Next, you add this operator as a task in your DAG. When the DAG runs, Airflow triggers the PythonOperator, which calls your function. The function executes its code, such as printing a message, and Airflow captures this output in logs. After the function finishes without errors, Airflow marks the task as successful. This process allows you to run any Python code as part of your Airflow workflows.

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