Azure operators in Apache Airflow - Commands & Configuration

Introduction

Azure operators in Airflow help you connect and interact with Microsoft Azure services easily. They let you automate tasks like creating virtual machines, uploading files to storage, or running Azure functions without writing complex code.

When you want to automate uploading data files to Azure Blob Storage as part of a workflow.

When you need to start or stop Azure virtual machines on a schedule.

When you want to trigger an Azure Function from an Airflow task.

When you need to run a data pipeline that reads from or writes to Azure Data Lake.

When you want to manage Azure resources like databases or containers automatically.

Config File - azure_dag.py

azure_dag.py

from airflow import DAG
from airflow.providers.microsoft.azure.operators.wasb_delete_blob import WasbDeleteBlobOperator
from airflow.providers.microsoft.azure.operators.wasb_upload import WasbUploadOperator
from airflow.utils.dates import days_ago

with DAG(dag_id='azure_blob_example', start_date=days_ago(1), schedule_interval='@daily', catchup=False) as dag:
    upload_blob = WasbUploadOperator(
        task_id='upload_blob',
        container_name='example-container',
        blob_name='data/sample.txt',
        file_path='/tmp/sample.txt',
        azure_blob_conn_id='azure_blob_default'
    )

    delete_blob = WasbDeleteBlobOperator(
        task_id='delete_blob',
        container_name='example-container',
        blob_name='data/old_sample.txt',
        azure_blob_conn_id='azure_blob_default'
    )

    upload_blob >> delete_blob

This Airflow DAG file defines a simple workflow using Azure Blob Storage operators.

WasbUploadOperator uploads a file to a specified Azure Blob container.

WasbDeleteBlobOperator deletes a specified blob from the container.

The DAG runs daily starting from yesterday and connects to Azure using the connection ID azure_blob_default.

Commands

List all available DAGs to verify that the Azure DAG is recognized by Airflow.

Terminal

airflow dags list

Expected Output

azure_blob_example

example_dag

Trigger the Azure Blob Storage DAG manually to start the upload and delete tasks.

Terminal

airflow dags trigger azure_blob_example

Expected Output

Created <DagRun azure_blob_example @ 2024-06-01T12:00:00+00:00: manual__2024-06-01T12:00:00+00:00, externally triggered: True>

List all tasks in the Azure Blob Storage DAG to see the upload and delete tasks.

Terminal

airflow tasks list azure_blob_example

Expected Output

upload_blob

delete_blob

Run the upload_blob task for the given date to test uploading a file to Azure Blob Storage.

Terminal

airflow tasks test azure_blob_example upload_blob 2024-06-01

Expected Output

[2024-06-01 12:00:00,000] {taskinstance.py:876} INFO - Executing <Task(WasbUploadOperator): upload_blob> on 2024-06-01

[2024-06-01 12:00:01,000] {wasb.py:123} INFO - Uploading file /tmp/sample.txt to container example-container as data/sample.txt

[2024-06-01 12:00:02,000] {taskinstance.py:1050} INFO - Task upload_blob succeeded

Key Concept

If you remember nothing else from Azure operators in Airflow, remember: they let you automate Azure service tasks easily within your workflows using simple, ready-made operators.

Common Mistakes

Not setting up the Azure connection in Airflow before running the operators.

The operators need credentials to access Azure services; without the connection, tasks will fail authentication.

Create an Azure connection in Airflow UI or CLI with the correct credentials and use its connection ID in the operator.

Using incorrect container or blob names in the operator parameters.

Azure Blob Storage will reject invalid names or non-existing containers, causing task failures.

Double-check container and blob names exist and follow Azure naming rules before running the DAG.

Summary

Define Azure operators in an Airflow DAG to automate tasks like uploading and deleting blobs.

Use airflow CLI commands to list, trigger, and test DAGs and tasks.

Ensure Azure connections are configured properly for authentication.

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