Creating custom operators in Apache Airflow - Performance & Efficiency
When we create custom operators in Airflow, we want to know how their work time changes as the tasks they handle grow.
We ask: How does the time to run the operator change when it processes more data or steps?
Analyze the time complexity of the following custom operator code snippet.
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults
class MyCustomOperator(BaseOperator):
@apply_defaults
def __init__(self, items, *args, **kwargs):
super().__init__(*args, **kwargs)
self.items = items
def execute(self, context):
for item in self.items:
self.process_item(item)
def process_item(self, item):
# Simulate processing
pass
This operator processes a list of items one by one during execution.
Look for loops or repeated steps in the code.
- Primary operation: The
forloop inexecutethat processes each item. - How many times: Once for every item in the
itemslist.
As the number of items grows, the operator runs process_item more times.
| Input Size (n) | Approx. Operations |
|---|---|
| 10 | 10 calls to process_item |
| 100 | 100 calls to process_item |
| 1000 | 1000 calls to process_item |
Pattern observation: The work grows directly with the number of items; doubling items doubles work.
Time Complexity: O(n)
This means the time to run the operator grows in a straight line with the number of items it processes.
[X] Wrong: "The operator runs in constant time no matter how many items there are."
[OK] Correct: Because the operator loops through each item, more items mean more work and more time.
Understanding how your custom operator scales helps you design efficient workflows and shows you can think about performance in real projects.
"What if process_item itself had a loop over a list inside it? How would the time complexity change?"
Practice
Solution
Step 1: Understand what custom operators do
Custom operators let you create tasks that run your own code inside Airflow workflows.Step 2: Identify where custom logic is placed
This logic is placed inside theexecutemethod of the operator class.Final Answer:
To define reusable tasks with custom logic inside theexecutemethod -> Option DQuick Check:
Custom operator = reusable task logic [OK]
- Confusing UI or scheduler settings with operators
- Thinking operators install plugins
- Believing custom operators change Airflow core behavior
Solution
Step 1: Recall Python class syntax for inheritance
In Python, to inherit from a class, useclass ClassName(BaseClass):.Step 2: Match Airflow custom operator syntax
Airflow custom operators inherit fromBaseOperator, so the correct syntax isclass MyOperator(BaseOperator):.Final Answer:
class MyOperator(BaseOperator): -> Option BQuick Check:
Python class inheritance = class MyOperator(BaseOperator): [OK]
- Using function definition instead of class
- Omitting BaseOperator inheritance
- Using incorrect syntax like 'operator' keyword
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'Solution
Step 1: Understand the execute method output
Theexecutemethod prints the string with prefix "Message:" followed by themessageattribute.Step 2: Substitute the message value
The task runs withmessage='Hello Airflow', so the print outputs "Message: Hello Airflow".Final Answer:
Message: Hello Airflow -> Option AQuick Check:
execute prints 'Message: ' + message [OK]
- Forgetting the 'Message:' prefix in print
- Confusing attribute name with string literal
- Assuming execute method is missing
from airflow.models import BaseOperator
class BadOperator(BaseOperator):
def __init__(self, param):
self.param = param
def execute(self, context):
print(self.param)Solution
Step 1: Check __init__ method for BaseOperator inheritance
When inheriting BaseOperator, you must callsuper().__init__(**kwargs)to initialize the base class properly.Step 2: Identify missing super() call
The code does not callsuper().__init__()</>, so Airflow internal setup will fail.Final Answer:
Missing call to super().__init__() in __init__ method -> Option CQuick Check:
Always call super().__init__() in custom operator init [OK]
- Omitting super().__init__() call
- Removing context parameter from execute
- Misunderstanding inheritance rules
Solution
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, callssuper().__init__(**kwargs), and stores the list inself.files.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 overself.filesand prints each filename, matching the requirement.Final Answer:
Option A correctly implements the custom operator to print each filename -> Option AQuick Check:
Use self.files list and loop in execute() [OK]
- Not calling super().__init__()
- Using wrong parameter types
- Referencing variables without self.
- Missing context parameter in execute
Ready to prove it?
Pass the test and this pattern moves unknown
