Ingestimento dati da un'API nelle pipeline

Ingerire da un'API significa estrarre dati tramite HTTP da un servizio web, solitamente come JSON paginato, invece di leggere da un file o da un database. A differenza dei file o di un message bus, non c'è una sorgente API generica integrata, quindi gestisci tu stesso autenticazione, paginazione e limiti di velocità. Le pipeline di Lakeflow supportano tre pattern per l'acquisizione da un'API qualsiasi. Quale si adatta dipende dal volume e dalle tue esigenze di refresh.

Importante

Prima di scrivere qualsiasi codice personalizzato per l'ingestione API, verifica se esiste già un connettore gestito per il tuo sorgente. Lakeflow Connect include connettori integrati per molte API SaaS (software as a service) comuni, come Salesforce, Workday, ServiceNow e Google Analytics, e c'è anche un numero crescente di connettori partner. Se è disponibile un connettore per la tua sorgente, gestisce per te l'autenticazione, la paginazione e l'estrazione incrementale, e quasi sempre richiede meno lavoro rispetto a un'ingestione implementata manualmente. Vedi i concetti relativi al connettore Lakeflow Connect. Usa i pattern qui sotto solo quando non si adatta alcun connettore.

Prerequisiti

  • Un oleodotto. Per crearne uno, consulta i tutorial sulle pipeline Lakeflow.
  • Le credenziali API, come un token o una chiave, memorizzate come segreto Azure Databricks. Mai codificare manualmente le credenziali nel codice sorgente della pipeline. Vedere Gestione dei segreti.
  • Accesso di rete dalle risorse di calcolo della pipeline all’endpoint dell’API.
  • Familiarità con le tabelle di streaming e le viste materializzate, i tipi di dataset che questi modelli producono. Vedi Tabelle in streaming e Visualizzazioni materializzate.

Scegliere un modello

Non esiste una fonte REST-API generica nativa nelle pipeline, quindi quando prelevi da un'API arbitraria, scegli uno dei tre pattern in base al volume dei dati e alla frequenza con cui assumi:

Modello Usa quando
Ritratti periodici come vista materializzata I payload sono piccoli o medi e vengono prelevati una volta per ogni esecuzione di pipeline, come dati di riferimento, tassi valutari giornalieri o un'API paginata ma limitata.
API Python Data Source Devi interrogare periodicamente un'API ad alto volume o di streaming in modo incrementale, con l'avanzamento salvato tramite checkpoint, in modo che, in caso di riavvio, non venga riletto tutto.
Ingestione disaccoppiata con Auto Loader Vuoi isolare le particolarità specifiche delle API dalla logica di trasformazione e ottenere senza alcuno sforzo aggiuntivo il tracciamento dei file una e una sola volta.

Pattern 1: Ritiri periodici come vista materializzata

Per payload di dimensioni medio-piccole recuperati una volta per ogni esecuzione della pipeline, scrivi una funzione Python che chiami l'API e restituisca un DataFrame Spark. Poiché il dataset è una vista materializzata, la pipeline riesegue la funzione integralmente e in modo idempotente ogni volta che si aggiorna.

I seguenti passaggi illustrano come creare una vista materializzata con aggiornamenti periodici:

  1. Memorizza il token API in un segreto, poi mappalo a una proprietà di configurazione Spark nelle impostazioni della pipeline così il codice della pipeline può leggerlo. Aggiungi la proprietà al blocco spark_conf della configurazione del cluster della pipeline:

    {
  "clusters": [
    {
      "spark_conf": {
        "api.token": "{{secrets/<scope-name>/<secret-name>}}"
      }
    }
  ]
}

    Il codice nel passo successivo legge questo valore con spark.conf.get("api.token"). Per maggiori informazioni sulla configurazione dei segreti nelle impostazioni della pipeline, vedi Accesso sicuro alle credenziali di archiviazione con segreti in una pipeline.

  2. Definiamo una vista materializzata che chiama l'API e restituisce la risposta come DataFrame:

    import requests
from pyspark import pipelines as dp
from pyspark.sql import Row

@dp.materialized_view(
    name="exchange_rates_bronze",
    comment="Daily FX rates pulled from a public REST API",
)
def exchange_rates_bronze():
    resp = requests.get(
        "https://api.example.com/v1/rates",
        params={"base": "USD"},
        headers={"Authorization": f"Bearer {spark.conf.get('api.token')}"},
        timeout=30,
    )
    resp.raise_for_status()
    rates = resp.json()["rates"]
    rows = [Row(currency=k, rate=float(v), as_of_date=resp.json()["date"]) for k, v in rates.items()]
    return spark.createDataFrame(rows)

  3. Gestisci la paginazione all'interno della funzione facendo un loop sulle pagine e concatenando i risultati prima di restituire il DataFrame:

    import requests
from pyspark import pipelines as dp
from pyspark.sql import Row

@dp.materialized_view(
    name="customers_bronze",
    comment="Customers pulled from a paginated REST API",
)
def customers_bronze():
    token = spark.conf.get("api.token")
    rows = []
    url = "https://api.example.com/v1/customers"
    while url:  # follow the API's next-page cursor until exhausted
        resp = requests.get(
            url,
            headers={"Authorization": f"Bearer {token}"},
            timeout=30,
        )
        resp.raise_for_status()
        payload = resp.json()
        rows.extend(Row(**record) for record in payload["data"])
        url = payload.get("next")  # None on the last page
    return spark.createDataFrame(rows)

    Aggiungi logica di ritentativi e di retrocesso attorno alla richiesta di resilienza.

Questo pattern rilegge l'intera risposta API ad ogni aggiornamento della pipeline, quindi usalo solo quando il payload è limitato. Per le letture incrementali, utilizza il modello 2.

Pattern 2: API ad alto volume o streaming con l'API Data Source Python

Per le API per cui devi eseguire query incrementali con tracciamento dell'offset, devi implementare una sorgente dati personalizzata usando l'API Python per le origini dati di Spark. Questo ti garantisce una corretta semantica di streaming, incluso l'avanzamento salvato tramite checkpoint e le letture incrementali, in modo che, dopo un riavvio, l'elaborazione riprenda dall'ultimo offset invece di interrogare di nuovo l'intera API.

I seguenti passaggi ti mostrano come ingerire da una fonte di dati personalizzata:

  1. Implementare un DataSource e DataSourceStreamReader che chiamano l'API e tengono traccia dell'offset di lettura. Per dettagli sulla creazione di una fonte di dati personalizzata, vedi PySpark fonti di dati personalizzate.

  2. Registra la sorgente dati in modo che la pipeline possa fare riferimento a essa per nome del formato:

    spark.dataSource.register(MyApiDataSource)

  3. Leggi dalla fonte registrata in una tabella di streaming:

    from pyspark import pipelines as dp

@dp.table(name="events_bronze")
def events_bronze():
    return spark.readStream.format("my_api_source").load()

Schema 3: Scollegare l'ingestione da un lavoro programmato e da Auto Loader

Un modello di produzione comune è separare la chiamata API dalla pipeline. Un lavoro programmato invia le risposte API grezze come file in un volume del Catalogo Unity, e la pipeline le rileva con Auto Loader. Questo isola le peculiarità specifiche dell’API, come la paginazione e i limiti di richiesta, dalla tua logica di trasformazione dichiarativa e ti offre automaticamente il tracciamento dei file exactly-once di Auto Loader.

I seguenti passaggi ti mostrano come separare l'ingestione da un lavoro programmato:

  1. Scrivi un notebook o uno script che chiami l'API e scriva le risposte JSON non elaborate in un volume di Unity Catalog. Leggi le credenziali API da un segreto. Vedere Gestione dei segreti.

    import requests, json, time

token = dbutils.secrets.get(scope="<scope-name>", key="<secret-name>")
volume_path = "/Volumes/main/raw/landing/api_events"

resp = requests.get(
    "https://api.example.com/v1/events",
    headers={"Authorization": f"Bearer {token}"},
    timeout=30,
)
resp.raise_for_status()
# One file per run; the pipeline's Auto Loader tracks which files it has ingested.
with open(f"{volume_path}/events_{int(time.time())}.json", "w") as f:
    json.dump(resp.json()["data"], f)

  2. Programma il notebook o lo script per farlo girare autonomamente con Lakeflow Jobs. Consulta Attività di Lakeflow.

  3. Nella tua pipeline, definisci una tabella di streaming che legga i file landed con Auto Loader:

    from pyspark import pipelines as dp

@dp.table(name="api_events_bronze")
def api_events_bronze():
    return (
        spark.readStream.format("cloudFiles")
            .option("cloudFiles.format", "json")
            .load("/Volumes/main/raw/landing/api_events")
    )

Per saperne di più sull'ingestione affidabile di file con Auto Loader, vedi Carica file dall'archiviazione cloud degli oggetti e Cos'è Auto Loader?.

Migliori pratiche per l'ingestione di API

  • Tieni i segreti fuori dal codice sorgente. Memorizza token API e chiavi in scope segreti di Azure Databricks e leggili in runtime. Vedere Gestione dei segreti.
  • Valida le risposte in anticipo. Aggiungi aspettative sulle righe ingerite per individuare risposte API malformate prima che fluiscano a valle.
  • Gestisci la paginazione e i limiti di tariffa. Scorri le pagine e aggiungi un meccanismo di ritentativo con backoff, in modo che un errore temporaneo non comprometta l'intero aggiornamento.

Risorse aggiuntive

  • Last updated on