Erfassen von Daten mit Auto Loader

Wenn neue Datendateien kontinuierlich im Cloudspeicher ankommen, benötigen Sie eine effiziente Möglichkeit, sie zu verarbeiten, ohne manuell nachzuverfolgen, welche Dateien aufgenommen wurden. Auto Loader bewältigt diese Herausforderung, indem es neue Dateien automatisch erkennt und aufnimmt, sobald sie erscheinen, und so eine Exakt-einmal-Garantie bietet, ohne dass Sie den Status oder Prüfpunkte manuell verwalten müssen.

Grundlegendes zur Funktionsweise des Auto Loaders

Auto Loader ist eine strukturierte Streamingquelle, die einen Cloudspeicherort überwacht und neue Dateien inkrementell verarbeitet. Anstatt ganze Verzeichnisse wiederholt zu scannen, behält auto Loader den Zustand bei, welche Dateien sie bereits verarbeitet hat, und stellt sicher, dass jede Datei genau einmal aufgenommen wird.

Wenn Sie einen Stream für das automatische Laden starten, kann er vorhandene Dateien im Verzeichnis verarbeiten und dann kontinuierlich auf Neueinkünfte achten. Der Datenstrom speichert Fortschrittsinformationen an einem Prüfpunkt, sodass er genau an der Stelle fortgesetzt werden kann, an der er angehalten hat, wenn er unterbrochen wird.

Auto Loader erkennt neue Dateien mit einem von zwei Modi:

• Verzeichnisauflistungsmodus: Das automatische Laden listet regelmäßig das Eingabeverzeichnis auf, um neue Dateien zu ermitteln. Für diesen Ansatz ist keine zusätzliche Konfiguration erforderlich, die über den Speicherzugriff hinausgeht.
• Dateibenachrichtigungsmodus: Das automatische Laden verwendet Cloudbenachrichtigungsdienste, um Ereignisse zu empfangen, wenn neue Dateien eingehen. Dieser Modus ist für große Workloads effizienter, da wiederholte Verzeichnisauflistungen vermieden werden.

Für die meisten produktiven Arbeitslasten empfiehlt Databricks den Dateibenachrichtigungsmodus mit aktivierten Dateiereignissen für Ihren externen Speicherort in Unity Catalog. Mit Dateiereignissen erhält Auto Loader direkt Benachrichtigungen, wenn Dateien eintreffen, was die Latenzzeit und die Kosten für die Cloud-API reduziert.

Konfigurieren des automatischen Ladevorgangs für die Dateiaufnahme

Zum Einlesen von Daten mit dem Auto Loader verwenden Sie das cloudFiles Format mit spark.readStream. Im folgenden Beispiel werden JSON-Dateien aus Azure Data Lake Storage gelesen und in eine Unity-Katalogtabelle geschrieben:

base_path = "abfss://container@storage.dfs.core.windows.net/autoloader/orders"
schema_path = f"{base_path}/schema"
checkpoint_path = f"{base_path}/checkpoint"

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", schema_path)
  .load("abfss://container@storage.dfs.core.windows.net/incoming/orders/")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable("sales.bronze.orders"))

Die cloudFiles.format Option gibt das Quelldateiformat an. Auto Loader unterstützt JSON-, CSV-, Parkett-, Avro-, ORC-, XML-, Text- und Binärdateien. Das cloudFiles.schemaLocation speichert das abgeleitete Schema, wodurch das automatische Laden Schemaänderungen im Laufe der Zeit verfolgen kann. Während der Schemaspeicherort und der Prüfpunktspeicherort identisch sein können, ermöglicht es Ihnen ihre Trennung, einen Datenstrom zurückzusetzen, ohne das abgeleitete Schema zu verlieren.

Die trigger(availableNow=True) Einstellung verarbeitet alle derzeit verfügbaren Dateien und stoppt dann. Dieses Muster eignet sich gut für geplante Batchaufträge. Zur kontinuierlichen Verarbeitung können Sie den Trigger weglassen oder zum Verarbeiten von Dateien in Intervallen verwenden trigger(processingTime="1 minute") .

Verwenden der SQL-Syntax mit read_files

Auto Loader unterstützt auch die SQL-Syntax über die read_files Tabellenwertfunktion. Dieser Ansatz ist nützlich, wenn Sie SQL bevorzugen oder in SQL-Notizbüchern arbeiten:

SELECT * FROM read_files(
  'abfss://container@storage.dfs.core.windows.net/incoming/orders/',
  format => 'json',
  schemaHints => 'order_id INT, amount DECIMAL(10,2)'
)

Bei Verwendung von read_files mit Streamingtabellen in Lakeflow-Pipelines werden die Auto Loader-Funktionen automatisch aktiviert. Verwenden Sie das STREAM Schlüsselwort, um die inkrementelle Verarbeitung zu aktivieren:

CREATE OR REFRESH STREAMING TABLE bronze_orders
AS SELECT * FROM STREAM read_files(
  'abfss://container@storage.dfs.core.windows.net/incoming/orders/',
  format => 'json'
)

In diesem Zusammenhang verwaltet die Pipeline das Checkpointing und die Schemaentwicklung automatisch.

Verwaltung von Schemainferenz und Evolution

Das automatische Laden kann das Schema Ihrer Quelldateien automatisch erkennen, ohne dass Schemas manuell definiert werden müssen. Wenn die Schema-Inferenz aktiviert ist, prüft der Auto Loader Dateien, um Spaltennamen und Datentypen zu bestimmen.

Bei JSON- und CSV-Dateien leitet auto Loader standardmäßig alle Spalten als Zeichenfolgen ab. Dieser konservative Ansatz verhindert Typenkonflikte, die zu Datenverlusten führen könnten. Um die automatische Typerkennung zu aktivieren, legen Sie die cloudFiles.inferColumnTypes Option fest:

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.inferColumnTypes", "true")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load(source_path))

Wenn Ihre Quelldaten neue Spalten einführen, erkennt auto Loader die Änderung und kann das Zielschema automatisch weiterentwickeln. Die cloudFiles.schemaEvolutionMode Option steuert dieses Verhalten:

Die gerettete Datenspalte (_rescued_data) erfasst alle Daten, die nicht mit dem erwarteten Schema übereinstimmen, einschließlich Typenkonflikten und unerwarteten Spalten. Dieses Feature verhindert Datenverlust, wenn Quelldaten nicht den Erwartungen entsprechen. Auto Loader fügt diese Spalte automatisch hinzu, wenn das Schema abgeleitet wird.

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .option("cloudFiles.schemaEvolutionMode", "rescue")
  .schema(expected_schema)
  .load(source_path))

Verwenden Sie die rescuedDataColumn Option, um die Spalte für die geretteten Daten umzubenennen:

.option("rescuedDataColumn", "my_rescued_data")

Wenn Sie wissen, dass bestimmte Spalten bestimmte Typen aufweisen sollten, verwenden Sie Schemahinweise , um die abgeleiteten Typen außer Kraft zu setzen:

.option("cloudFiles.schemaHints", "order_date DATE, amount DECIMAL(10,2)")

Überwachen und Optimieren von Auto Loader-Datenströmen

Bei Produktionsworkloads hilft Ihnen die Überwachung beim Nachverfolgen des Erfassungsfortschritts und beim Identifizieren von Problemen. Auto Loader stellt eine SQL-Funktion bereit, um den Status der ermittelten Dateien abzufragen:

SELECT * FROM cloud_files_state('/path/to/checkpoint');

Diese Abfrage gibt Metadaten zu Dateien zurück, die das automatische Laden ermittelt hat, einschließlich des Verarbeitungsstatus. Sie können diese Informationen verwenden, um zu überprüfen, ob Dateien wie erwartet aufgenommen werden.

Metriken des Auto Loaders werden über das Status-Dashboard für strukturiertes Streaming gemeldet, einschließlich:

• numFilesOutstanding: Gefundene, aber noch nicht verarbeitete Dateien
• numBytesOutstanding: Gesamtanzahl der Bytes, die auf die Verarbeitung warten

Steuern Sie bei Workloads mit hohem Volumen die Verarbeitungsrate mithilfe cloudFiles.maxFilesPerTrigger oder cloudFiles.maxBytesPerTrigger. Diese Einstellungen verhindern, dass einzelne Mikrobatches zu groß werden:

.option("cloudFiles.maxFilesPerTrigger", "1000")
.option("cloudFiles.maxBytesPerTrigger", "1g")

Um auf Metadaten auf Dateiebene in Ihren aufgenommenen Daten zuzugreifen, wählen Sie die _metadata Spalte aus. Diese ausgeblendete Spalte enthält Informationen zu jeder Quelldatei, einschließlich file_path, , file_name, file_sizeund file_modification_time:

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "parquet")
  .load(source_path)
  .select("*", "_metadata.file_path", "_metadata.file_modification_time"))

Auto Loader lässt sich nahtlos in Lakeflow-Pipelines integrieren, wo es Prüfpunktsetzung und Schemaverwaltung automatisch übernimmt. In der nächsten Lektion erfahren Sie, wie Sie auto Loader in deklarativen Pipelinedefinitionen für End-to-End-Datenaufnahmeworkflows verwenden.