Daten aus einer API in Pipelines einbinden

Das Aufnehmen von einer API bedeutet, Daten über HTTP aus einem Webservice zu beziehen, meist als paginiertes JSON, anstatt aus einer Datei oder Datenbank zu lesen. Im Gegensatz zu Dateien oder einem Nachrichtenbus gibt es keine integrierte generische API-Quelle, sodass du Authentifizierung, Paginierung und Ratenbegrenzungen selbst erledigst. Lakeflow-Pipelines unterstützen drei Muster zur Aufnahme von einer beliebigen API. Welche passt, hängt von deinem Lautstärke- und Refresh-Bedarf ab.

Voraussetzungen

• Eine Pipeline. Um eines zu erstellen, siehe die Lakeflow-Pipelines-Tutorials.
• API-Zugangsdaten, wie ein Token oder Schlüssel, gespeichert als Azure Databricks Secret. Fügen Sie Anmeldeinformationen niemals fest in den Quellcode der Pipeline ein. Siehe Verwaltung von Geheimnissen.
• Netzwerkzugang von deiner Pipeline-Berechnung zum API-Endpunkt.
• Vertrautheit mit Streaming-Tabellen und materialisierten Ansichten sowie mit den Datensatztypen, die durch diese Muster hervorgebracht werden. Siehe Streaming-Tabellen und materialisierte Ansichten.

Auswählen eines Musters

Es gibt keine native generische REST-API Quelle in Pipelines, also wähle beim Ziehen aus einer beliebigen API eines von drei Mustern basierend auf dem Datenvolumen und der Häufigkeit, die du einführst:

Muster 1: Periodische Abrufe als materialisierte Ansicht

Für kleine bis mittlere Nutzlasten, die einmal pro Pipeline-Ausführung gezogen werden, schreiben Sie eine Python-Funktion, die die API aufruft und einen Spark DataFrame zurückgibt. Da es sich bei dem Dataset um eine materialisierte Ansicht handelt, führt die Pipeline die Funktion bei jeder Aktualisierung vollständig und idempotent erneut aus.

Die folgenden Schritte zeigen Ihnen, wie Sie eine materialisierte Ansicht mit regelmäßigen Abrufen erstellen können:

1. Speichere das API-Token in einem Geheimnis und ordne es dann einer Spark-Konfigurationseigenschaft in deinen Pipeline-Einstellungen zu, damit der Pipeline-Code es lesen kann. Füge die Eigenschaft dem spark_conf Block der Clusterkonfiguration der Pipeline hinzu:

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

   Der Code im nächsten Schritt liest diesen Wert mit spark.conf.get("api.token")aus. Mehr zur Konfiguration von Geheimnissen in Pipeline-Einstellungen finden Sie unter Sicherer Zugriff auf Speicherzugangsdaten mit Geheimnissen in einer Pipeline.

2. Definiere eine materialisierte Ansicht, die die API aufruft und die Antwort als DataFrame zurückgibt:

   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. Führen Sie die Paginierung innerhalb der Funktion durch, indem Sie Seiten durchlaufen und die Ergebnisse verketten, bevor Sie den DataFrame zurückgeben:

   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)
   

   Fügen Sie eine Logik für Wiederholung und Rückzug rund um die Bitte um Resilienz hinzu.

Dieses Muster liest die vollständige API-Antwort bei jedem Pipeline-Update erneut, daher wird es nur verwendet, wenn die Nutzlast begrenzt ist. Verwenden Sie für inkrementelle Lesevorgänge Muster 2.

Muster 2: APIs mit hohem Volumen oder Streaming mit der Python Data Source API

Für APIs musst du inkrementell mit Offset-Tracking abfragen und eine benutzerdefinierte Datenquelle mit Sparks Python Data Source API implementieren. Dadurch erhalten Sie eine korrekte Streaming-Semantik, einschließlich Checkpoint-basiertem Fortschritt und inkrementellen Lesevorgängen, sodass ein Neustart ab dem letzten Offset fortgesetzt wird, anstatt die gesamte API erneut zu pullen.

Die folgenden Schritte zeigen Ihnen, wie Sie aus einer benutzerdefinierten Datenquelle einnehmen:

1. Implementiere DataSource und DataSourceStreamReader, die die API aufrufen und den Leseoffset verfolgen. Für Details zur Erstellung einer benutzerdefinierten Datenquelle siehe PySpark benutzerdefinierte Datenquellen.

2. Registrieren Sie die Datenquelle, damit die Pipeline sie nach Formatnamen referenzieren kann:

   spark.dataSource.register(MyApiDataSource)
   

3. Lesen Sie aus der registrierten Quelle in eine Streaming-Tabelle ein:

   from pyspark import pipelines as dp
   
   @dp.table(name="events_bronze")
   def events_bronze():
       return spark.readStream.format("my_api_source").load()
   

Muster 3: Entkoppelte Datenaufnahme mit einem geplanten Job und dem Auto Loader

Ein gängiges Produktionsmuster ist es, den API-Aufruf von der Pipeline zu trennen. Ein geplanter Job speichert die rohen API-Antworten als Dateien in einem Unity Catalog-Volume, und die Pipeline übernimmt diese mit dem Auto Loader. Das isoliert API-spezifische Eigenheiten wie Paginierung und Ratenbegrenzungen von deiner deklarativen Transformationslogik und bietet dir die exakt einmalige Dateiverfolgung von Auto Loader kostenlos.

Die folgenden Schritte zeigen Ihnen, wie Sie die Datenaufnahme mit einem geplanten Job entkoppeln:

1. Schreibe ein Notizbuch oder Skript, das die API aufruft und die rohen JSON-Antworten auf ein Unity-Catalog-Volume schreibt. Lesen Sie die API-Anmeldeinformationen aus einem Secret aus. Siehe Verwaltung von Geheimnissen.

   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. Plane das Notizbuch oder Skript so, dass es eigenständig mit Lakeflow-Jobs läuft. Siehe Lakeflow Jobs.

3. Definiere in deiner Pipeline eine Streaming-Tabelle, die die gelandeten Dateien mit Auto Loader liest:

   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")
       )
   

Mehr über zuverlässige Dateiaufnahme mit Auto Loader finden Sie unter Dateien aus Cloud-Objektspeicher laden und Was ist Auto Loader?.

Best Practices für API-Aufnahme

• Geheimnisse aus dem Quellcode heraushalten. Speichern Sie API-Token und Schlüssel in Azure Databricks Secret-Bereichen und lesen Sie sie zur Laufzeit aus. Siehe Verwaltung von Geheimnissen.
• Validiere Antworten frühzeitig. Füge Erwartungen auf die aufgenommenen Zeilen hinzu, um fehlerhafte API-Antworten zu erkennen, bevor sie stromabwärts fließen.
• Behandeln Sie Paginierung und Ratenbeschränkungen. Durchlaufen Sie die Seiten in einer Schleife und fügen Sie Wiederholungsversuche mit Backoff hinzu, damit ein vorübergehender Fehler nicht die gesamte Aktualisierung zum Scheitern bringt.

Weitere Ressourcen

• Laden von Daten in Pipelines
• Lakeflow Connect-Verbindungskonzepte
• PySpark benutzerdefinierte Datenquellen
• Verwalten Sie die Datenqualität anhand von Pipeline-Erwartungen