Fabric Lakehouse - Load data - Apache Airflow Load Control

Applies to
Platform: Fabric Lakehouse

Load a deployed Fabric Lakehouse solution with Apache Airflow by running the DAG that the generator writes. The DAG has one task per loader and runs the tasks in dependency order, one at a time. Afterwards every loader has run once.

The DAG is available since Generator 2.2. With Target Platform Fabric, each task starts a loader notebook through the Fabric REST API. With Target Platform Azure Synapse, each task runs a Synapse pipeline, see Azure Synapse.

Prerequisites

LoadControl Options contains Airflow, and Configure LoadControl is true. The artifacts then contain LoadControl\<project>_dag.py.
The artifacts are deployed without errors.
An Apache Airflow job in the Fabric workspace of the deployment. Its requirements contain the Python packages msal and requests, which the DAG imports.
The client ID, a client secret and the tenant ID of the service principal that is a Member of the workspace, see Fabric Lakehouse - Target environment.

Run the DAG

In replacement_config.json, check the variable airflow#airflow#fabric_conn_id. It names the Airflow connection of the DAG and is fabric-integration by default. Then replace the placeholders.
In the Apache Airflow job, create a connection with this connection ID. Enter the client ID as login, the client secret as password, and {"tenantId": "<tenant ID>"} as extra.
Run .\deploy.ps1 as described in Fabric Lakehouse - Deployment. After the notebooks, the script uploads the DAG file to the first Apache Airflow job of the workspace. It fills in the workspace ID and the ID of each loader notebook. It does this also with -SkipDeployment.
In Airflow, unpause the DAG and trigger it. The DAG ID is <project>, the project name in lower case with underscores, for example test_vaultandmart_spark_fabric. The DAG has no schedule and is paused when it is created.

Each task gets an access token for the service principal with msal and starts its loader notebook as a notebook job. It then reads the job status every 30 seconds until the job completes or fails. Every task passes the start of the data interval of the DAG run as load timestamp, so all loaders of one run share it.

Check the result

The DAG run succeeded when all its tasks have the state success. The log of each loader task ends with Completed successfully!. The loader notebooks write their row counts to the execution log, as in Fabric Lakehouse - Load data - Native Load Control.

Example

For the project test.vaultandmart.spark.fabric, the DAG test_vaultandmart_spark_fabric holds this task for the Stage loader. deploy.ps1 replaces <notebook_id> with the ID of the notebook STG_ST_BDC_CUSTOMER_Loader:

run_stg_st_bdc_customer_loader = PythonOperator(
    task_id="STG_ST_BDC_CUSTOMER_Loader",
    python_callable=run_fabric_job_and_wait,
    op_kwargs={
        'item_id': '<notebook_id>',
        'load_timestamp':
          "{{ data_interval_start.strftime(
             '%Y-%m-%d %H:%M:%S.%f') }}"
    }
)

The line breaks inside the template are added here for the page width.

Azure Synapse

With Target Platform Azure Synapse, the DAG runs one Synapse pipeline per loader with the operator AzureDataFactoryRunPipelineOperator. The artifacts contain the pipeline definitions as LoadControl\<loader>_Pipeline.json; the deployment notebooks do not create them. The DAG uses the Airflow connection ID AzureDataFactory, which is written into the DAG. It retries a failed task three times and passes the logical date of the DAG run as load timestamp. It uses arguments of Apache Airflow 2, such as schedule_interval and concurrency.

In replacement_config.json, enter the Spark pool as datafactory_pipeline#datafactory_pipeline#sparkpool_name and the linked service as datafactory_pipeline#datafactory_pipeline#linked_service_reference_name. Then replace the placeholders.
Create the pipelines in the Synapse workspace from the files <loader>_Pipeline.json.
Create the Airflow connection AzureDataFactory with the provider apache-airflow-providers-microsoft-azure.
Copy LoadControl\<project>_dag.py into the DAG folder of Airflow, then unpause and trigger the DAG.

The other pipeline variables are preset: timeout 0.12:00:00, retry 0, retry_interval_in_seconds 30, secure_input and secure_output false.

Failure symptoms and recovery

The deployment script finds no Apache Airflow job

Message: No Airflow job found in workspace <workspace name>
Cause: The workspace has no Apache Airflow job. The script stops after the deployment.
Fix: Create an Apache Airflow job in the workspace and run .\deploy.ps1 -SkipDeployment.

A task fails before the notebook starts

Message: Token error: <description> or Start failed: <response>
Cause: The connection holds a wrong client ID, secret or tenant ID, or the service principal has no access to the workspace.
Fix: Correct the connection or the workspace access, then clear the task and its downstream tasks in Airflow.

A loader notebook fails

Message: Fabric job failed with status: Failed
Cause: The loader notebook failed; its downstream tasks do not run.
Fix: Read the ERROR rows of the execution log, correct the cause, and clear the task and its downstream tasks.