Creating custom operators in Apache Airflow - Quick Revision & Summary

Recall & Review

beginner

What is a custom operator in Apache Airflow?

A custom operator is a user-defined class that extends Airflow's BaseOperator to perform specific tasks not covered by built-in operators.

beginner

Which method must be implemented when creating a custom operator?

The execute() method, which contains the code that runs when the operator is triggered.

intermediate

How do you pass parameters to a custom operator?

Parameters are passed via the operator's constructor and stored as instance variables to be used in the execute() method.

intermediate

Why should you call super().__init__() in a custom operator's constructor?

To properly initialize the BaseOperator part of the class, ensuring Airflow can manage the operator's lifecycle and features.

beginner

What is the benefit of creating custom operators in Airflow?

They allow you to reuse complex or specific task logic easily across multiple DAGs, improving code organization and maintainability.

What base class should you extend to create a custom operator in Airflow?

A BaseOperator

B DAG

C TaskInstance

D PythonOperator

Custom operators extend BaseOperator to inherit core operator functionality.

Which method contains the code that runs when a custom operator executes?

A run()

B execute()

C start()

D trigger()

The execute() method is called by Airflow to run the operator's task.

Why is calling super().__init__() important in a custom operator?

A To run the task

B To initialize the DAG

C To start the scheduler

D To initialize the BaseOperator

It ensures the BaseOperator is properly initialized with Airflow's internal setup.

How do you pass custom parameters to your operator?

A Via the operator's constructor

B By setting environment variables

C Through the execute() method

D Using Airflow variables only

Parameters are passed when creating the operator instance via its constructor.

What is a main advantage of using custom operators?

A They reduce the number of DAGs

B They replace the need for sensors

C They allow reuse of task logic across DAGs

D They speed up the scheduler

Custom operators help reuse complex task logic, making DAGs cleaner and easier to maintain.

Explain the steps to create a custom operator in Airflow.

Think about class inheritance and method overriding.

Describe why and when you would create a custom operator instead of using built-in ones.

Consider task uniqueness and code reuse.

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