PythonOperator for custom logic in Apache Airflow - Commands & Configuration

Introduction

Sometimes you need to run your own Python code as part of a workflow. PythonOperator lets you run custom Python functions inside Airflow tasks to automate any logic you want.

When you want to run a Python function that processes data as part of your workflow.

When you need to call an API or perform calculations inside a task.

When you want to reuse existing Python code in your Airflow pipeline.

When you want to run simple scripts without creating separate executables.

When you want to chain multiple Python tasks with dependencies.

Config File - example_dag.py

example_dag.py

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

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

def add_numbers(a, b):
    print(f'Sum is {a + b}')

def my_dag():
    with DAG(dag_id='custom_python_operator_dag', start_date=datetime(2024, 1, 1), schedule_interval='@daily', catchup=False) as dag:
        task1 = PythonOperator(
            task_id='greet_task',
            python_callable=greet
        )
        task2 = PythonOperator(
            task_id='add_task',
            python_callable=add_numbers,
            op_kwargs={'a': 5, 'b': 7}
        )
        task1 >> task2
    return dag

dag = my_dag()

This file defines an Airflow DAG named custom_python_operator_dag that runs daily starting from January 1, 2024.

It has two tasks using PythonOperator:

• greet_task: runs the greet function which prints a greeting.
• add_task: runs add_numbers with arguments 5 and 7, printing their sum.

The tasks run in order: greet_task first, then add_task.

Commands

List all available DAGs to verify that your DAG file is recognized by Airflow.

Terminal

airflow dags list

Expected Output

custom_python_operator_dag

example_other_dag

Manually trigger the DAG to run your PythonOperator tasks immediately.

Terminal

airflow dags trigger custom_python_operator_dag

Expected Output

Created <DagRun custom_python_operator_dag @ 2024-06-01T12:00:00+00:00: manual__2024-06-01T12:00:00+00:00, externally triggered: True>

List all tasks in the DAG to see the task IDs you can monitor or run individually.

Terminal

airflow tasks list custom_python_operator_dag

Expected Output

greet_task

add_task

Run the greet_task for the given date to test the Python function without running the whole DAG.

Terminal

airflow tasks test custom_python_operator_dag greet_task 2024-06-01

Expected Output

[2024-06-01 12:00:00,000] {taskinstance.py:1234} INFO - Hello from PythonOperator!

[2024-06-01 12:00:00,001] {taskinstance.py:1235} INFO - Task greet_task succeeded

Key Concept

If you remember nothing else from this pattern, remember: PythonOperator lets you run any Python function as a task inside Airflow workflows.

Common Mistakes

Passing the function call instead of the function itself to python_callable (e.g., python_callable=greet()).

This runs the function immediately when the DAG is parsed, not when the task runs, causing errors or unexpected behavior.

Pass the function name without parentheses: python_callable=greet.

Not setting op_kwargs or op_args when the Python function requires parameters.

The task will fail because the function expects arguments but none are provided.

Use op_kwargs={'param': value} or op_args=[value] to pass arguments to the function.

Not setting catchup=False in the DAG when testing manually.

Airflow tries to run all past scheduled runs, which can be confusing during development.

Set catchup=False in the DAG definition to avoid running old schedules.

Summary

Define Python functions with the logic you want to run as tasks.

Use PythonOperator in your DAG to run these functions as tasks.

Trigger and test tasks using Airflow CLI commands to verify behavior.

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