Trasformare i dati con dbt

2025-04-21

Nota

Il processo Apache Airflow è alimentato da Apache Airflow.

dbt(Data Build Tool) è un'interfaccia della riga di comando open source che semplifica la trasformazione e la modellazione dei dati all'interno dei data warehouse gestendo codice SQL complesso in modo strutturato e gestibile. Consente ai team di dati di creare trasformazioni affidabili e testabili al centro delle pipeline analitiche.

In combinazione con Apache Airflow, le funzionalità di trasformazione di dbt vengono migliorate dalle funzionalità di pianificazione, orchestrazione e gestione delle attività di Airflow. Questo approccio combinato, usando l'esperienza di trasformazione di dbt insieme alla gestione del flusso di lavoro di Airflow, offre pipeline di dati efficienti e affidabili, portando infine a decisioni più rapide e dettagliate basate sui dati.

Questa esercitazione illustra come creare un daG Apache Airflow che usa dbt per trasformare i dati archiviati in Microsoft Fabric Data Warehouse.

Prerequisiti

Per iniziare, è necessario soddisfare i prerequisiti seguenti:

Creare il principale del servizio. Aggiungere il principale servizio come Contributor nell'area di lavoro in cui si crea il data warehouse.
Se non ne hai uno, Crea un magazzino Fabric. Inserire i dati di esempio nel warehouse usando la pipeline di dati. Per questa esercitazione si userà l'esempio NYC Taxi-Green.
Creare il "processo Apache Airflow" nell'area di lavoro.

Trasformare i dati archiviati in Fabric Warehouse usando dbt

La sezione guida l'utente nell'esecuzione dei passaggi seguenti:

Specificare i requisiti.
Creare un progetto dbt nell'archiviazione gestita di Fabric fornita dal processo Apache Airflow.
Creare un DAG Apache Airflow per orchestrare i processi dbt

Specificare i requisiti

Creare un file requirements.txt nella cartella dags. Aggiungere i pacchetti seguenti come requisiti di Apache Airflow.

astronomer-cosmos: questo pacchetto viene usato per eseguire i progetti core dbt come DAG di Apache Airflow e gruppi di task.
dbt-fabric: questo pacchetto viene usato per creare un progetto dbt, che può quindi essere distribuito in un data warehouse di Fabric
```
astronomer-cosmos==1.10.1
dbt-fabric==1.9.5   
```

Creare un progetto dbt nell'archiviazione gestita di Fabric fornita dal processo Apache Airflow.

In questa sezione viene creato un progetto dbt di esempio nel processo Apache Airflow per il set di dati nyc_taxi_green con la struttura di directory seguente.

   dags
   |-- my_cosmos_dag.py
   |-- nyc_taxi_green
   |  |-- profiles.yml
   |  |-- dbt_project.yml
   |  |-- models
   |  |   |-- nyc_trip_count.sql
   |  |-- target

Creare la cartella denominata nyc_taxi_green nella cartella dags con il file profiles.yml. Questa cartella contiene tutti i file necessari per il progetto dbt.

Copiare i contenuti seguenti in profiles.yml. Questo file di configurazione contiene i dettagli e i profili di connessione del database usati da dbt. Aggiornare i valori segnaposto e salvare il file.

config:
  partial_parse: true
nyc_taxi_green:
  target: fabric-dev
  outputs:
    fabric-dev:
      type: fabric
      driver: "ODBC Driver 18 for SQL Server"
      server: <sql connection string of your data warehouse>
      port: 1433
      database: "<name of the database>"
      schema: dbo
      threads: 4
      authentication: ServicePrincipal
      tenant_id: <Tenant ID of your service principal>
      client_id: <Client ID of your service principal>
      client_secret: <Client Secret of your service principal>

Creare il file dbt_project.yml e copiare il contenuto seguente nel file. Questo file specifica la configurazione a livello di progetto.

name: "nyc_taxi_green"

config-version: 2
version: "0.1"

profile: "nyc_taxi_green"

model-paths: ["models"]
seed-paths: ["seeds"]
test-paths: ["tests"]
analysis-paths: ["analysis"]
macro-paths: ["macros"]

target-path: "target"
clean-targets:
  - "target"
  - "dbt_modules"
  - "logs"

require-dbt-version: [">=1.0.0", "<2.0.0"]

models:
  nyc_taxi_green:
    materialized: table

Creare la cartella models nella cartella nyc_taxi_green. Per questa esercitazione viene creato il modello di esempio nel file denominato nyc_trip_count.sql che crea la tabella che mostra il numero di corse al giorno per fornitore. Copiare il contenuto seguente nel file.

   with new_york_taxis as (
       select * from nyctlc
   ),
   final as (
     SELECT
       vendorID,
       CAST(lpepPickupDatetime AS DATE) AS trip_date,
       COUNT(*) AS trip_count
     FROM
         [contoso-data-warehouse].[dbo].[nyctlc]
     GROUP BY
         vendorID,
         CAST(lpepPickupDatetime AS DATE)
     ORDER BY
         vendorID,
         trip_date;
   )
   select * from final

Creare un DAG Apache Airflow per orchestrare i processi dbt

Creare il file denominato my_cosmos_dag.py nella dags cartella e incollarvi il contenuto seguente.

import os
from pathlib import Path
from datetime import datetime
from cosmos import DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig
from airflow import DAG

DEFAULT_DBT_ROOT_PATH = Path(__file__).parent.parent / "dags" / "nyc_taxi_green"
DBT_ROOT_PATH = Path(os.getenv("DBT_ROOT_PATH", DEFAULT_DBT_ROOT_PATH))
profile_config = ProfileConfig(
     profile_name="nyc_taxi_green",
     target_name="fabric-dev",
     profiles_yml_filepath=DBT_ROOT_PATH / "profiles.yml",
)

dbt_fabric_dag = DbtDag(
     project_config=ProjectConfig(DBT_ROOT_PATH,),
     operator_args={"install_deps": True},
     profile_config=profile_config,
     schedule_interval="@daily",
     start_date=datetime(2023, 9, 10),
     catchup=False,
     dag_id="dbt_fabric_dag",
)

Esegui il tuo DAG

Esegui il DAG all'interno del job di Apache Airflow.
Per visualizzare il DAG caricato nell'interfaccia utente di Apache Airflow, fare clic su Monitor in Apache Airflow.

Convalida i tuoi dati

Dopo un'esecuzione corretta, per convalidare i dati, è possibile visualizzare la nuova tabella denominata "nyc_trip_count.sql" creata nel data warehouse di Fabric.

Guida introduttiva: Creare un processo Apache Airflow

Condividi tramite