Gegevens invoeren van een API in pijplijnen

Invoeren van een API betekent dat je data via HTTP uit een webservice haalt, meestal als gepagineerde JSON, in plaats van uit een bestand of database te lezen. In tegenstelling tot bestanden of een berichtenbus is er geen ingebouwde generieke API-bron, dus je regelt authenticatie, paginering en snelheidslimieten zelf. Lakeflow-pipelines ondersteunen drie patronen voor het opnemen van gegevens uit een willekeurige API. Welke optie het beste past, hangt af van het volume dat je nodig hebt en je verversingsbehoeften.

Belangrijk

Voordat je aangepaste API-invoercode schrijft, controleer je of er al een managed connector bestaat voor je broncode. Lakeflow Connect levert ingebouwde connectors voor veel gangbare software as a service (SaaS) API's, zoals Salesforce, Workday, ServiceNow en Google Analytics, en er is ook een groeiend aantal partnerconnectors. Als een connector je bron dekt, regelt hij authenticatie, paginering en incrementele extractie voor je, en dat is bijna altijd minder werk dan een handgerolde ingestie. Zie concepten van Lakeflow Connect-connectors. Gebruik de onderstaande patronen alleen als er geen connector past.

Prerequisites

  • Een pijpleiding. Om er een te maken, zie de tutorials van Lakeflow pipelines.
  • API-inloggegevens, zoals een token of sleutel, opgeslagen als een Azure Databricks geheim. Codeer nooit inloggegevens hardcode in de broncode van de pijplijn. Zie Geheimbeheer.
  • Netwerktoegang van je pijplijn-berekening naar het API-eindpunt.
  • Vertrouwdheid met streamingtabellen en gematerialiseerde weergaven, en met de typen datasets die deze patronen voortbrengen. Zie Streamingtabellen en gematerialiseerde weergaven.

Een patroon kiezen

Er is geen native generieke REST-API bron in pipelines, dus wanneer je uit een willekeurige API haalt, kies dan een van drie patronen op basis van datavolume en hoe vaak je het invoert:

Patroon Wanneer gebruiken
Periodieke pulls als een gematerialiseerd beeld Payloads zijn klein tot middelgroot en worden eenmaal per pipeline-run gehaald, zoals referentiedata, dagelijkse FX-koersen of een gepagineerde maar begrensde API.
Python Data Source API Je moet een API met een hoog volume of een streaming-API incrementeel bevragen, waarbij de voortgang via checkpoints wordt opgeslagen zodat na een herstart niet alles opnieuw hoeft te worden ingelezen.
Ontkoppelde inname met Auto Loader Je wilt API-specifieke eigenaardigheden isoleren uit je transformatielogica en exact één bestandstracking gratis krijgen.

Patroon 1: Periodieke ophalingen als een gematerialiseerde weergave

Voor kleine tot middelgrote payloads die één keer per pijplijn worden opgehaald, schrijf een Python-functie die de API aanroept en een Spark DataFrame teruggeeft. Omdat de dataset een gematerialiseerde weergave is, voert de pijplijn de functie telkens volledig en idempotent opnieuw uit wanneer de pijplijn wordt bijgewerkt.

De volgende stappen laten zien hoe je een gematerialiseerd beeld kunt bouwen met periodieke trekken:

  1. Sla het API-token op in een geheim en koppel het vervolgens aan een Spark-configuratie-eigenschap in je pipeline-instellingen zodat de pipeline-code het kan lezen. Voeg de eigenschap toe aan het spark_conf blok van de clusterconfiguratie van de pijplijn:

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

    De code in de volgende stap leest deze waarde met spark.conf.get("api.token"). Voor meer informatie over het configureren van geheimen in pipeline-instellingen, zie Veilig toegang krijgen tot opslaggegevens met geheimen in een pipeline.

  2. Definieer een gematerialiseerde weergave die de API aanroept en het antwoord teruggeeft als een 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. Behandel de paginering binnen de functie door over pagina's heen te lopen en de resultaten te koppelen voordat je het DataFrame terugstuurt:

    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)

    Voeg herproberen- en terugtreklogica toe rond het verzoek om veerkracht.

Dit patroon leest de volledige API-respons opnieuw bij elke pipeline-update, dus gebruik het alleen wanneer de payload begrensd is. Voor incrementele leesbewerkingen gebruik je patroon 2.

Patroon 2: API's met een hoog volume of streaming-API's met de Python Data Source API

Voor API's waarvoor je incrementeel en met het bijhouden van offsets periodiek gegevens moet opvragen, moet je een aangepaste gegevensbron implementeren met de Python Data Source API van Spark. Dit zorgt voor juiste streamingsemantiek, inclusief via checkpoints opgeslagen voortgang en incrementele leesbewerkingen, zodat een herstart vanaf de laatste offset hervat in plaats van de volledige API opnieuw op te vragen.

De volgende stappen laten zien hoe je kunt invoeren uit een aangepaste databron:

  1. Implementeer een DataSource en DataSourceStreamReader die de API aanroepen en de leesoffset bijhouden. Voor details over het authoren van een aangepaste databron, zie PySpark aangepaste databronnen.

  2. Registreer de databron zodat de pipeline deze kan verwijzen op formaatnaam:

    spark.dataSource.register(MyApiDataSource)

  3. Lees uit de geregistreerde bron in een streaming-tabel:

    from pyspark import pipelines as dp

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

Patroon 3: Ontkoppel gegevensinname met een geplande taak en Auto Loader

Een veelvoorkomend productiepatroon is het scheiden van de API-aanroep van de pipeline. Een geplande taak landt de ruwe API-antwoorden als bestanden in een Unity Catalog-volume, en de pipeline pikt ze op met Auto Loader. Dit isoleert API-specifieke eigenaardigheden zoals paginering en snelheidslimieten van je declaratieve transformatielogica, en geeft je Auto Loader gratis de exakt-één bestandsregistratie.

De volgende stappen laten zien hoe je inname kunt loskoppelen van een geplande klus:

  1. Schrijf een notitieboek of script dat de API aanroept en de ruwe JSON-antwoorden naar een Unity Catalog-volume schrijft. Lees de API-inloggegevens van een geheim exemplaar. Zie Geheimbeheer.

    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. Plan het notitieboek of script zo in dat het zelfstandig draait met Lakeflow Jobs. Zie Lakeflow Jobs.

  3. Definieer in je pipeline een streamingtabel die de gelandde bestanden leest met 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")
    )

Voor meer informatie over betrouwbare bestandsinvoer met Auto Loader, zie Bestanden laden uit cloud object storage en Wat is Auto Loader?.

Best practices voor API-ingestie

  • Houd geheimen buiten de broncode. Sla API-tokens en sleutels op in Azure Databricks geheime scopes en lees ze tijdens runtime. Zie Geheimbeheer.
  • Valideer antwoorden vroeg. Voeg verwachtingen toe op de ingelezen rijen om misvormde API-antwoorden te vangen voordat ze stroomafwaarts stromen.
  • Pak paginatiek en snelheidslimieten aan. Loop over pagina's en voeg opnieuw proberen toe met backoff zodat een tijdelijke storing niet de hele update laat mislukken.

Aanvullende bronnen

  • Last updated on