PythonOperator for custom logic in Apache Airflow - Interactive Code Practice

Practice - 5 Tasks

1fill in blank

easy

Complete the code to import the PythonOperator from Airflow.

Apache Airflow

from airflow.operators.[1] import PythonOperator

Drag options to blanks, or click blank then click option'

Apython

Bdummy

Cemail

Dbash

Attempts:

3 left

💡 Hint

Common Mistakes

Importing from the wrong operator module like 'bash' or 'email'.

Misspelling 'python' in the import statement.

The PythonOperator is imported from airflow.operators.python.

2fill in blank

medium

Complete the code to define a Python function named 'my_task' that prints 'Hello Airflow'.

Apache Airflow

def [1]():
    print('Hello Airflow')

Drag options to blanks, or click blank then click option'

Ahello

Brun_task

Ctask_func

Dmy_task

Attempts:

3 left

💡 Hint

Common Mistakes

Using a different function name than 'my_task'.

Forgetting to define a function.

The function must be named my_task as referenced later in the operator.

3fill in blank

hard

Fix the error in the PythonOperator task by completing the argument that specifies the Python callable.

Apache Airflow

task = PythonOperator(
    task_id='print_hello',
    python_callable=[1],
    dag=dag
)

Drag options to blanks, or click blank then click option'

Aprint('my_task')

Bmy_task()

Cmy_task

D'my_task'

Attempts:

3 left

💡 Hint

Common Mistakes

Passing the function call with parentheses.

Passing the function name as a string.

The python_callable argument expects the function name without parentheses or quotes.

4fill in blank

hard

Fill both blanks to create a DAG with id 'my_dag' and schedule interval '@daily'.

Apache Airflow

dag = DAG(
    dag_id=[1],
    schedule_interval=[2]
)

Drag options to blanks, or click blank then click option'

A'my_dag'

B'@hourly'

C'@daily'

D'default_dag'

Attempts:

3 left

💡 Hint

Common Mistakes

Using the wrong schedule interval like '@hourly'.

Not using quotes around strings.

The DAG id should be 'my_dag' and the schedule interval '@daily' for daily runs.

5fill in blank

hard

Fill all three blanks to create a PythonOperator named 'task' with task_id 'print_hello', using the function 'my_task' and the DAG 'dag'.

Apache Airflow

task = [1](
    task_id=[2],
    python_callable=[3],
    dag=dag
)

Drag options to blanks, or click blank then click option'

APythonOperator

B'print_hello'

Cmy_task

DBashOperator

Attempts:

3 left

💡 Hint

Common Mistakes

Using BashOperator instead of PythonOperator.

Passing task_id without quotes.

Passing python_callable as a string or with parentheses.

The operator must be PythonOperator, the task_id a string 'print_hello', and the callable the function my_task.

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