Creating custom operators in Apache Airflow - Practice Exercises

Challenge - 5 Problems

🧠 Conceptual

intermediate

Purpose of Custom Operators in Airflow

Why would you create a custom operator in Apache Airflow instead of using built-in operators?

A To encapsulate reusable logic specific to your workflows that built-in operators do not provide.

B Because built-in operators cannot be used in DAGs.

C To avoid writing any Python code in your DAG files.

D To automatically speed up task execution without any configuration.

💡 Hint

Think about when you need to do something unique or repetitive in your workflows.

Custom operators let you package specific task logic that is not covered by Airflow's built-in operators, making your DAGs cleaner and more reusable.

💻 Command Output

intermediate

Output of Custom Operator Execution

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

Apache Airflow

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

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

    def execute(self, context):
        print(f"Hello, {self.name}!")

# Assume this operator is used in a DAG with name='Airflow'

A No output printed

B Hello, Airflow!

C Error: __init__ missing required positional argument

D Hello, name!

💡 Hint

Look at how the name parameter is used in execute.

The operator prints the greeting using the name passed during initialization, so it prints 'Hello, Airflow!'.

❓ Configuration

advanced

Correct Custom Operator Initialization

Which option correctly defines a custom operator that accepts a file_path parameter and passes other arguments properly to the base class?

A

class FileOperator(BaseOperator):
    def __init__(self, file_path, **kwargs):
        super().__init__(**kwargs)
        self.file_path = file_path

B

class FileOperator(BaseOperator):
    def __init__(self, **kwargs):
        self.file_path = kwargs.pop('file_path')
        super().__init__(**kwargs)

C

class FileOperator(BaseOperator):
    def __init__(self, file_path, **kwargs):
        self.file_path = file_path
        super().__init__(**kwargs)

D

class FileOperator(BaseOperator):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.file_path = kwargs['file_path']

💡 Hint

Remember to call super().__init__ before using self attributes.

Option A correctly calls super().__init__(**kwargs) first and then assigns file_path. This is the recommended pattern.

❓ Troubleshoot

advanced

Error Raised by Incorrect Custom Operator Code

What error will this custom operator code raise when instantiated?

Apache Airflow

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

A RuntimeError: Task instance failed

B AttributeError: 'BadOperator' object has no attribute 'param'

C TypeError: __init__() missing 1 required positional argument: 'task_id'

D No error, runs fine

💡 Hint

Check the order of super().__init__ call and attribute assignment.

Because super().__init__ is called after assigning self.param, Airflow's BaseOperator may not initialize properly, causing a TypeError about missing arguments.

🔀 Workflow

expert

Order of Steps to Create and Use a Custom Operator

What is the correct order of steps to create and use a custom operator in Airflow?

A 3,1,2,4

B 2,1,3,4

C 1,3,2,4

D 1,2,3,4

💡 Hint

Think about writing code, importing it, then using it in the DAG.

First, define the operator class. Then import it in your DAG file. Next, create an instance with parameters. Finally, add it to the DAG's task flow.