Azure operators in Apache Airflow - Step-by-Step Execution

Process Flow - Azure operators

Start DAG

↓

Trigger Azure Operator

↓

Authenticate with Azure

↓

Execute Azure Task

↓

Check Task Status

↓

Success?

No→Handle Failure

Yes↓

Complete DAG

This flow shows how an Airflow DAG triggers an Azure operator, which authenticates, runs a task on Azure, checks the result, and completes or handles failure.

Execution Sample

Apache Airflow

from airflow.providers.microsoft.azure.operators.adls import AzureDataLakeStorageCreateDirectoryOperator

create_dir = AzureDataLakeStorageCreateDirectoryOperator(
    task_id='create_dir',
    directory_name='myfolder',
    azure_data_lake_conn_id='azure_default'
)

This code creates an Airflow task that uses an Azure operator to create a directory in Azure Data Lake Storage.

Process Table

Step	Action	Azure Operator Method	Input Parameters	Result/Output
1	Start DAG run	N/A	N/A	DAG started
2	Trigger Azure operator task	execute	directory_name='myfolder', azure_data_lake_conn_id='azure_default'	Operator initialized
3	Authenticate with Azure	get_conn	azure_data_lake_conn_id='azure_default'	Authentication successful
4	Create directory in ADLS	create_directory	directory_name='myfolder'	Directory 'myfolder' created
5	Check operation status	check_status	N/A	Success
6	Mark task complete	N/A	N/A	Task succeeded
7	DAG completes	N/A	N/A	DAG run finished successfully

💡 DAG run ends after successful Azure operator task execution and directory creation.

Status Tracker

Variable	Start	After Step 2	After Step 3	After Step 4	After Step 5	Final
directory_name	N/A	'myfolder'	'myfolder'	'myfolder'	'myfolder'	'myfolder'
conn_id	N/A	'azure_default'	'azure_default'	'azure_default'	'azure_default'	'azure_default'
task_status	N/A	initialized	authenticated	directory created	success	success

Key Moments - 3 Insights

Why does the operator need a connection ID like 'azure_default'?

The connection ID tells the operator how to authenticate with Azure. Without it, the operator cannot access Azure services. See execution_table step 3 where authentication happens using 'azure_default'.

What happens if the directory already exists?

The operator may raise an error or skip creation depending on its implementation. This would cause the task to fail at step 4 or 5 in the execution_table, triggering failure handling.

Is the DAG run complete immediately after triggering the operator?

No, the DAG waits for the operator to finish its task successfully (step 6) before completing (step 7). This ensures the Azure task finished properly.

Visual Quiz - 3 Questions

Look at the execution_table, what is the result after step 4?

A Authentication successful

B Directory 'myfolder' created

C Task succeeded

D DAG started

💡 Hint

Check the 'Result/Output' column for step 4 in the execution_table.

At which step does the operator authenticate with Azure?

A Step 3

B Step 5

C Step 2

D Step 6

💡 Hint

Look for 'Authentication successful' in the Result/Output column.

If the connection ID was incorrect, which step would fail?

A Step 6

B Step 4

C Step 3

D Step 7

💡 Hint

Authentication happens at step 3; failure there means no access to Azure.

Concept Snapshot

Azure operators in Airflow:
- Use Azure connection ID for authentication
- Operators perform Azure tasks (e.g., create directory)
- DAG waits for operator success before completing
- Errors in Azure tasks cause DAG failure
- Common operators: ADLS, VM, Blob Storage
- Define tasks with operator classes and parameters

Full Transcript

This visual execution shows how an Airflow DAG uses an Azure operator to create a directory in Azure Data Lake Storage. The DAG starts, triggers the operator, which authenticates using a connection ID, performs the directory creation, checks the result, and then completes the DAG run. Variables like directory_name and task_status change as the operator progresses. Key points include the need for correct Azure authentication and that the DAG waits for the operator to finish before completing. The quiz questions reinforce understanding of the steps and outcomes.

Practice

1. What is the main purpose of Azure operators in Apache Airflow?

easy

A. To automate and manage Azure cloud services within Airflow workflows

B. To create user interfaces for Azure services

C. To replace Azure CLI commands manually

D. To monitor local server hardware performance

Solution

1. Step 1: Understand Azure operators role

   Azure operators are designed to automate tasks on Azure cloud services from Airflow.

2. Step 2: Compare options with this role

   Only To automate and manage Azure cloud services within Airflow workflows describes automation and management of Azure services, which matches the purpose.

3. Final Answer:

   To automate and manage Azure cloud services within Airflow workflows -> Option A

4. Quick Check:

   Azure operators automate Azure tasks = C [OK]

Hint: Azure operators automate Azure tasks in Airflow [OK]

Common Mistakes:

• Confusing operators with UI tools
• Thinking operators replace manual CLI commands only
• Assuming operators monitor local hardware

2. Which of the following is the correct way to import the Azure Data Factory operator in Airflow?

easy

A. import AzureDataFactoryRunPipelineOperator from airflow.microsoft.azure

B. from airflow.providers.microsoft.azure.operators.data_factory import AzureDataFactoryRunPipelineOperator

C. from airflow.operators.azure.datafactory import RunPipelineOperator

D. from airflow.providers.azure.operators.datafactory import AzureRunPipelineOperator

Solution

1. Step 1: Recall correct import path for Azure Data Factory operator

   The official import path uses 'airflow.providers.microsoft.azure.operators.data_factory' and the class name 'AzureDataFactoryRunPipelineOperator'.

2. Step 2: Check each option for syntax and correctness

   from airflow.providers.microsoft.azure.operators.data_factory import AzureDataFactoryRunPipelineOperator matches the correct module and class name exactly; others have wrong paths or class names.

3. Final Answer:

   from airflow.providers.microsoft.azure.operators.data_factory import AzureDataFactoryRunPipelineOperator -> Option B

4. Quick Check:

   Correct import path and class name = A [OK]

Hint: Use full provider path with correct class name [OK]

Common Mistakes:

• Using incomplete or wrong import paths
• Incorrect class names or module names
• Mixing provider namespaces

3. Given this Airflow task using Azure VM operator:

run_vm = AzureVirtualMachineStartOperator(
    task_id='start_vm',
    resource_group_name='mygroup',
    vm_name='myvm'
)

What will happen when this task runs?

medium

A. It will create a new virtual machine named 'myvm'

B. It will stop the Azure virtual machine named 'myvm' in 'mygroup'

C. It will start the Azure virtual machine named 'myvm' in 'mygroup'

D. It will delete the virtual machine named 'myvm'

Solution

1. Step 1: Identify operator purpose

   The AzureVirtualMachineStartOperator is designed to start an existing VM.

2. Step 2: Analyze parameters and action

   The task specifies resource group and VM name, so it will start that VM, not stop, create, or delete it.

3. Final Answer:

   It will start the Azure virtual machine named 'myvm' in 'mygroup' -> Option C

4. Quick Check:

   Start operator starts VM = A [OK]

Hint: Operator name shows action: Start means start VM [OK]

Common Mistakes:

• Confusing start with stop or delete operators
• Assuming it creates a VM
• Ignoring resource group parameter

4. You wrote this Airflow task to delete an Azure storage container:

delete_container = AzureStorageDeleteOperator(
    task_id='delete_container',
    container_name='data',
    connection_string='my_conn_str'
)

But the task fails with an error about missing parameters. What is the likely cause?

medium

A. The operator does not support deleting containers, only blobs

B. The operator requires 'account_name' and 'account_key' instead of 'connection_string'

C. The operator requires 'resource_group_name' parameter

D. The operator requires the 'blob_name' parameter to delete a container

Solution

1. Step 1: Understand AzureStorageDeleteOperator capabilities

   This operator deletes blobs inside containers, not containers themselves.

2. Step 2: Analyze error cause

   Trying to delete a container causes failure because container deletion is unsupported by this operator.

3. Final Answer:

   The operator does not support deleting containers, only blobs -> Option A

4. Quick Check:

   Operator deletes blobs, not containers = D [OK]

Hint: Check operator docs for supported resource types [OK]

Common Mistakes:

• Assuming container deletion is supported
• Confusing connection parameters
• Adding unrelated parameters

5. You want to run an Azure Data Factory pipeline from Airflow and wait for its completion. Which operator and parameters combination is correct?

hard

A. AzureDataFactoryRunPipelineOperator with parameters: pipeline_id, resource_group, factory, async=True

B. AzureDataFactoryRunOperator with parameters: pipeline_name, resource_group_name, factory_name, wait_for_completion=True

C. AzureDataFactoryPipelineOperator with parameters: pipeline_name, resource_group_name, factory_name, wait=True

D. AzureDataFactoryRunPipelineOperator with parameters: pipeline_name, resource_group_name, factory_name, wait_for_termination=True

Solution

1. Step 1: Identify correct operator and parameters

   The official operator is AzureDataFactoryRunPipelineOperator. It uses 'pipeline_name', 'resource_group_name', 'factory_name', and 'wait_for_termination' to wait for completion.

2. Step 2: Compare options for exact names and flags

   AzureDataFactoryRunPipelineOperator with parameters: pipeline_name, resource_group_name, factory_name, wait_for_termination=True matches the correct operator and parameters exactly; others have wrong names or parameters.

3. Final Answer:

   AzureDataFactoryRunPipelineOperator with parameters: pipeline_name, resource_group_name, factory_name, wait_for_termination=True -> Option D

4. Quick Check:

   Correct operator and wait param = B [OK]

Hint: Use 'wait_for_termination=True' to wait for pipeline finish [OK]

Common Mistakes:

• Using wrong operator class names
• Incorrect parameter names
• Confusing async flags