Creating custom operators in Apache Airflow - Visual Walkthrough

Process Flow - Creating custom operators

Define Custom Operator Class

↓

Inherit from BaseOperator

↓

Override execute() Method

↓

Use Custom Operator in DAG

↓

Airflow Scheduler Runs DAG

↓

Custom Operator execute() Runs

↓

Task Completes or Fails

This flow shows how to create a custom operator by defining a class, overriding its execute method, then using it in a DAG where Airflow runs it.

Execution Sample

Apache Airflow

from airflow.models import BaseOperator

class MyOperator(BaseOperator):
    def execute(self, context):
        print('Hello from MyOperator')

Defines a custom operator class that prints a message when executed.

Process Table

Step	Action	Evaluation	Result
1	Define MyOperator class inheriting BaseOperator	Class created	Ready to use in DAG
2	Override execute() method	Method defined	Prints message when run
3	Instantiate MyOperator in DAG	Task created	Task ready for scheduling
4	Airflow scheduler triggers task	Task starts	execute() method runs
5	execute() prints message	Output: Hello from MyOperator	Task completes successfully
6	Task status updated	Success recorded	DAG continues or ends

💡 Task completes after execute() method runs and prints message

Status Tracker

Variable	Start	After Step 2	After Step 3	After Step 5	Final
MyOperator class	Not defined	Defined with execute()	Instance created	execute() runs	Task done
execute() output	None	Defined	None	'Hello from MyOperator'	Printed

Key Moments - 2 Insights

Why do we override the execute() method in the custom operator?

Because the execute() method contains the code that runs when the task executes, as shown in execution_table step 4 and 5.

Can we use the custom operator without inheriting BaseOperator?

No, inheriting BaseOperator is required to integrate with Airflow's task system, as shown in execution_table step 1.

Visual Quiz - 3 Questions

Look at the execution table, what happens at step 4?

A The Airflow scheduler triggers the task and execute() runs

B The custom operator class is defined

C The task status is updated to success

D The execute() method is overridden

💡 Hint

Refer to execution_table row with Step 4 describing task start and execute() running

According to variable_tracker, what is the output of execute() after step 5?

A None

B 'Hello from MyOperator'

C Task instance

D Class definition

💡 Hint

Check execute() output column after Step 5 in variable_tracker

If we do not inherit BaseOperator, what will happen according to the key moments?

A The task will run normally

B execute() will run twice

C Airflow will not recognize the custom operator as a task

D The DAG will fail to parse

💡 Hint

See key_moments about inheritance requirement for Airflow integration

Concept Snapshot

Creating custom operators in Airflow:
- Define a class inheriting BaseOperator
- Override execute(self, context) method
- Put task logic inside execute()
- Use the operator in DAGs like any other task
- Airflow runs execute() when task runs
- Enables reusable, custom task behavior

Full Transcript

To create a custom operator in Airflow, you start by defining a new class that inherits from BaseOperator. This inheritance is necessary so Airflow can treat your class as a task. Next, you override the execute() method in your class. This method contains the code that runs when the task executes. For example, you can print a message inside execute(). After defining your custom operator, you use it in your DAG by creating an instance of it as a task. When the Airflow scheduler runs your DAG, it triggers your custom operator's execute() method. The task runs, prints the message, and completes successfully. This process allows you to create reusable, custom task logic integrated with Airflow's scheduling and execution system.

Practice

1. What is the main purpose of creating a custom operator in Airflow?

easy

A. To install new Airflow plugins automatically

B. To change the Airflow webserver UI appearance

C. To configure the Airflow scheduler settings

D. To define reusable tasks with custom logic inside the execute method

Solution

1. Step 1: Understand what custom operators do

   Custom operators let you create tasks that run your own code inside Airflow workflows.

2. Step 2: Identify where custom logic is placed

   This logic is placed inside the execute method of the operator class.

3. Final Answer:

   To define reusable tasks with custom logic inside the execute method -> Option D

4. Quick Check:

   Custom operator = reusable task logic [OK]

Hint: Custom operators run your code in the execute() method [OK]

Common Mistakes:

• Confusing UI or scheduler settings with operators
• Thinking operators install plugins
• Believing custom operators change Airflow core behavior

2. Which of the following is the correct way to start defining a custom operator class in Airflow?

easy

A. operator MyOperator extends BaseOperator:

B. class MyOperator(BaseOperator):

C. class MyOperator():

D. def MyOperator():

Solution

1. Step 1: Recall Python class syntax for inheritance

   In Python, to inherit from a class, use class ClassName(BaseClass):.

2. Step 2: Match Airflow custom operator syntax

   Airflow custom operators inherit from BaseOperator, so the correct syntax is class MyOperator(BaseOperator):.

3. Final Answer:

   class MyOperator(BaseOperator): -> Option B

4. Quick Check:

   Python class inheritance = class MyOperator(BaseOperator): [OK]

Hint: Custom operators inherit BaseOperator with class syntax [OK]

Common Mistakes:

• Using function definition instead of class
• Omitting BaseOperator inheritance
• Using incorrect syntax like 'operator' keyword

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

from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults

class PrintOperator(BaseOperator):
    @apply_defaults
    def __init__(self, message: str, **kwargs):
        super().__init__(**kwargs)
        self.message = message

    def execute(self, context):
        print(f"Message: {self.message}")

# Task instance runs with message='Hello Airflow'

medium

A. Message: Hello Airflow

B. Hello Airflow

C. Message: message

D. Error: missing execute method

Solution

1. Step 1: Understand the execute method output

   The execute method prints the string with prefix "Message:" followed by the message attribute.

2. Step 2: Substitute the message value

   The task runs with message='Hello Airflow', so the print outputs "Message: Hello Airflow".

3. Final Answer:

   Message: Hello Airflow -> Option A

4. Quick Check:

   execute prints 'Message: ' + message [OK]

Hint: execute() prints 'Message: ' plus the message string [OK]

Common Mistakes:

• Forgetting the 'Message:' prefix in print
• Confusing attribute name with string literal
• Assuming execute method is missing

4. What is wrong with this custom operator code?

from airflow.models import BaseOperator

class BadOperator(BaseOperator):
    def __init__(self, param):
        self.param = param

    def execute(self, context):
        print(self.param)

medium

A. param should be a class variable, not instance variable

B. execute method should not have context parameter

C. Missing call to super().__init__() in __init__ method

D. BaseOperator cannot be inherited

Solution

1. Step 1: Check __init__ method for BaseOperator inheritance

   When inheriting BaseOperator, you must call super().__init__(**kwargs) to initialize the base class properly.

2. Step 2: Identify missing super() call

   The code does not call super().__init__()</>, so Airflow internal setup will fail.

3. Final Answer:

   Missing call to super().__init__() in __init__ method -> Option C

4. Quick Check:

   Always call super().__init__() in custom operator init [OK]

Hint: Always call super().__init__() in __init__ [OK]

Common Mistakes:

• Omitting super().__init__() call
• Removing context parameter from execute
• Misunderstanding inheritance rules

5. You want to create a custom operator that accepts a list of filenames and prints each filename during execution. Which code snippet correctly implements this behavior?

hard

A. class FilePrinterOperator(BaseOperator):
@apply_defaults
def __init__(self, files: list, **kwargs):
super().__init__(**kwargs)
self.files = files
def execute(self, context):
for f in self.files:
print(f)

B. class FilePrinterOperator(BaseOperator):
def __init__(self, files):
self.files = files
def execute(self):
for f in files:
print(f)

C. class FilePrinterOperator(BaseOperator):
@apply_defaults
def __init__(self, files: str, **kwargs):
super().__init__(**kwargs)
self.files = files
def execute(self, context):
print(self.files)

D. class FilePrinterOperator(BaseOperator):
@apply_defaults
def __init__(self, files: list, **kwargs):
super().__init__(**kwargs)
self.files = files
def execute(self, context):
print(files)

Solution

1. Step 1: Check constructor and inheritance

   class FilePrinterOperator(BaseOperator): @apply_defaults def __init__(self, files: list, **kwargs): super().__init__(**kwargs) self.files = files def execute(self, context): for f in self.files: print(f) correctly uses @apply_defaults, calls super().__init__(**kwargs), and stores the list in self.files.

2. Step 2: Verify execute method logic

   class FilePrinterOperator(BaseOperator): @apply_defaults def __init__(self, files: list, **kwargs): super().__init__(**kwargs) self.files = files def execute(self, context): for f in self.files: print(f) loops over self.files and prints each filename, matching the requirement.

3. Final Answer:

   Option A correctly implements the custom operator to print each filename -> Option A

4. Quick Check:

   Use self.files list and loop in execute() [OK]

Hint: Use self.files list and loop inside execute() [OK]

Common Mistakes:

• Not calling super().__init__()
• Using wrong parameter types
• Referencing variables without self.
• Missing context parameter in execute