PythonOperator for custom logic in Apache Airflow - Mini Project: Build & Apply

PythonOperator for custom logic

📖 Scenario: You are working with Apache Airflow to automate tasks. You want to create a simple workflow that runs a custom Python function using the PythonOperator.This project will guide you step-by-step to create a DAG with a PythonOperator that runs a function printing a greeting message.

🎯 Goal: Build an Airflow DAG that uses PythonOperator to run a custom Python function which prints a greeting message.

📋 What You'll Learn

Create a Python function called greet that prints 'Hello from Airflow!'

Create a DAG named greeting_dag with start_date set to January 1, 2024

Add a PythonOperator task named greet_task that calls the greet function

Set the task greet_task as the only task in the DAG

Print the task instance context in the greet function

💡 Why This Matters

🌍 Real World

Automating workflows in data pipelines, ETL jobs, or any scheduled tasks using Apache Airflow.

💼 Career

Understanding how to use PythonOperator is essential for DevOps engineers and data engineers working with Airflow to automate and manage workflows.

1

Create the greet function

Create a Python function called greet that takes a single argument **kwargs and prints the message 'Hello from Airflow!'.

Apache Airflow

# Your code here

Hint

Define a function named greet with **kwargs to accept context.

2

Create the Airflow DAG

Import DAG and datetime. Create a DAG named greeting_dag with start_date set to datetime(2024, 1, 1) and schedule_interval set to @daily.

Apache Airflow

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

# Your code here

Hint

Use DAG and datetime to create the DAG with the given name and schedule.

3

Add the PythonOperator task

Import PythonOperator from airflow.operators.python. Create a task named greet_task using PythonOperator that calls the greet function and belongs to the greeting_dag DAG.

Apache Airflow

from airflow import DAG
from datetime import datetime

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

from airflow.operators.python import PythonOperator

# Your code here

Hint

Create a PythonOperator with the specified task_id, python_callable, and dag.

4

Print the task instance context in greet and run the task

Modify the greet function to print the task_instance from kwargs using kwargs['ti']. Then add a line to print 'Task greet_task is set up in DAG greeting_dag'.

Apache Airflow

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

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

# Your code here

Hint

Use print(f"Task instance: {kwargs['ti']}") inside greet and print the confirmation message after task creation.

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