apache-airflow-microsoft-fabric-plugin

Apache Airflow Plugin for Microsoft Fabric Plugin. 🚀

Introduction

A Python package that helps Data and Analytics engineers trigger run on demand job items of Microsoft Fabric in Apache Airflow DAGs.

Microsoft Fabric is an end-to-end analytics and data platform designed for enterprises that require a unified solution. It encompasses data movement, processing, ingestion, transformation, real-time event routing, and report building. It offers a comprehensive suite of services including Data Engineering, Data Factory, Data Science, Real-Time Analytics, Data Warehouse, and Databases.

How to Use

Install the Plugin

Pypi package: https://pypi.org/project/apache-airflow-microsoft-fabric-plugin/

pip install apache-airflow-microsoft-fabric-plugin

Prerequisities

Before diving in,

• The plugin supports the authentication using user tokens. Tenant level admin account must enable the setting Allow user consent for apps. Refer to: Configure user consent
• Create a Microsoft Entra Id app if you don’t have one.
• You must have Refresh token.

Since custom connection forms aren't feasible in Apache Airflow plugins, use can use Generic connection type. Here's what you need to store:

• Connection Id: Name of the connection Id
• Connection Type: Generic
• Login: The Client ID of your service principal.
• Password: The refresh token fetched using Microsoft OAuth.
• Extra: { "tenantId": "The Tenant Id of your service principal", "clientSecret": "(optional) The Client Secret for your Entra ID App" "scopes": "(optional) Scopes you used to fetch the refresh token" }

NOTE: Default scopes applied are: https://api.fabric.microsoft.com/Item.Execute.All, https://api.fabric.microsoft.com/Item.ReadWrite.All, offline_access, openid, profile

Operators

FabricRunItemOperator

This operator composes the logic for this plugin. It triggers the Fabric item run and pushes the details in Xcom. It can accept the following parameters:

• workspace_id: The workspace Id.
• item_id: The Item Id. i.e Notebook and Pipeline.
• fabric_conn_id: Connection Id for Fabric.
• job_type: "RunNotebook" or "Pipeline".
• wait_for_termination: (Default value: True) Wait until the run item.
• timeout: int (Default value: 60 * 60 * 24 * 7). Time in seconds to wait for the pipeline or notebook. Used only if wait_for_termination is True.
• check_interval: int (Default value: 60s). Time in seconds to wait before rechecking the refresh status.
• max_retries: int (Default value: 5 retries). Max number of times to poll the API for a valid response after starting a job.
• retry_delay: int (Default value: 1s). Polling retry delay.
• deferrable: Boolean. Use the operator in deferrable mode.
• job_params: Dict. Parameters to pass into the job.

Features

• Refresh token rotation:

  Refresh token rotation is a security mechanism that involves replacing the refresh token each time it is used to obtain a new access token. This process enhances security by reducing the risk of stolen tokens being reused indefinitely.

• Xcom Integration:

  The Fabric run item enriches the Xcom with essential fields for downstream tasks:

  • run_id: Run Id of the Fabric item.
  • run_status: Fabric item run status.
    • In Progress: Item run is in progress.
    • Completed: Item run successfully completed.
    • Failed: Item run failed.
    • Disabled: Item run is disabled by a selective refresh.
  • run_location: The location of item run status.

• External Monitoring link:

  The operator conveniently provides a redirect link to the Microsoft Fabric item run.

• Deferable Mode:

  The operator runs in deferrable mode. The operator is deferred until the target status of the item run is achieved.

Sample DAG to use the plugin.

Ready to give it a spin? Check out the sample DAG code below:

from __future__ import annotations

from airflow import DAG
from apache_airflow_microsoft_fabric_plugin.operators.fabric import FabricRunItemOperator
from airflow.utils.dates import days_ago

default_args = {
    "owner": "airflow",
    "start_date": days_ago(1),
}

with DAG(
    dag_id="fabric_items_dag",
    default_args=default_args,
    schedule_interval="@daily",
    catchup=False,
) as dag:

    run_notebook = FabricRunItemOperator(
        task_id="run_fabric_notebook",
        workspace_id="<workspace_id>",
        item_id="<item_id>",
        fabric_conn_id="fabric_conn_id",
        job_type="RunNotebook",
        wait_for_termination=True,
        deferrable=True,
    )

    run_notebook

Feel free to tweak and tailor this DAG to suit your needs!

Contributing

We welcome any contributions:

• Report all enhancements, bugs, and tasks as GitHub issues
• Provide fixes or enhancements by opening pull requests in GitHub.