Bewährte Methoden für das automatische Laden

Auf dieser Seite werden bewährte Methoden beschrieben, die Sie anwenden können, um das automatische Ladeprogramm so zu konfigurieren, dass es zuverlässig, kosteneffizient und in großem Umfang für Ihren Anwendungsfall ausgeführt wird.

Diese bewährten Methoden reduzieren den Betriebsaufwand und verhindern häufige Probleme, die in der Produktion schwer zu diagnostizieren sind, z. B. unnötige LIST API-Kosten durch vollständige Verzeichnisscans, unbemerkten Datenverlust durch Schemadrift und durch eine Fehlkonfiguration des Prüfpunkts verursachte Pipeline-Neustarts.

Details zur Produktionskonfiguration finden Sie unter Konfigurieren des automatischen Ladevorgangs für Produktionsworkloads. Zur Überwachung und Beobachtbarkeit siehe Auto Loader überwachen und beobachten.

Auswählen des richtigen Ausführungsframeworks

Das beste Ausführungsframework für Ihren Anwendungsfall hängt davon ab, wie viel Kontrolle Sie über die Pipeline benötigen, und wie viel Betriebsaufwand Sie verwalten möchten. Für die meisten Benutzer und Produktionspipelines ist Auto Loader mit Lakeflow-Pipelines gut geeignet. Wenn Sie jedoch maximale Kontrolle und Anpassung benötigen, verwenden Sie auto Loader mit strukturiertem Streaming. Verwenden Sie für die einfachste Einrichtung mit einer verwalteten Oberfläche einen verwalteten LakeFlow-Connector, wenn verfügbar.

Lakeflow-Pipelines erweitern strukturiertes Streaming mit automatischer Skalierung, Datenqualitätsprüfungen, Schemaentwicklungsbehandlung und Überwachung über das Ereignisprotokoll. Databricks empfiehlt Lakeflow-Pipelines für die meisten Produktionsauslastungen.

Wählen Sie den richtigen Zeitplanungs- und Triggertyp aus

Der beste Zeitplanungs- und Triggertyp für Ihren Anwendungsfall hängt von Ihren Latenzanforderungen und Dateieingangsmustern ab. Für die meisten Anwendungsfälle empfiehlt Databricks einen Trigger bei Dateieingang, wenn Dateiereignisse aktiviert sind. Dies ermöglicht eine latenzarme Datenaufnahme zu geringen Kosten, da Rechenressourcen nur dann genutzt werden, wenn neue Dateien eintreffen. Die drei Triggertypen unterscheiden sich darin, wann und wie oft die Pipeline gestartet wird:

• Fortlaufend: Die Pipeline wird ohne Unterbrechung ausgeführt. Wird nur verwendet, wenn die Latenz von Unter-Sekunden eine harte Anforderung ist, da fortlaufende Berechnungskosten mehr kosten. Mit Dateiereignissen koppeln.
• Dateiankunftstrigger: Die Pipeline beginnt, wenn neue Dateien am Quellspeicherort landen. Optimal für niedrige bis mittlere Latenz oder unregelmäßige Dateieingangsmuster. Erfordert, dass Dateiereignisse aktiviert werden. Weitere Informationen finden Sie unter Auslösen von Aufträgen beim Eintreffen neuer Dateien.
• Geplant: Die Pipeline wird nach einem zeitbasierten Zeitplan ausgeführt (z. B. jede Stunde). Verwenden Sie dies, wenn die Latenzanforderungen nicht streng sind (Minuten bis Stunden). Funktioniert mit der Verzeichnisauflistung, aber Dateiereignisse senken die Kosten sogar im geplanten Modus, indem vollständige Verzeichnisüberprüfungen vermieden werden.

Ausführliche Informationen zur Verwendung Trigger.AvailableNow für die Batchplanung finden Sie unter "Using Trigger.AvailableNow" und "Rate limiting".

Auswählen des richtigen Modus für die Dateiermittlung

Auto Loader unterstützt drei Dateiermittlungsmodi mit unterschiedlichen Kompromissen bei der Einrichtungskomplexität, Skalierbarkeit und Kosten.

Dateiereignisse konsolidieren Cloudspeicher-Ressourcen, indem je ein Abonnement und eine Warteschlange pro externem Speicherort verwendet werden statt je eines pro Stream. Der Leistungsunterschied ist im großen Maßstab von Bedeutung: Die Verzeichnisauflistung muss das gesamte Quellverzeichnis für jeden Trigger überprüfen, sodass die Erfassungszeit mit der Verzeichnisgröße länger wird. Dateiereignisse liefern Benachrichtigungen über neue Dateien direkt, sodass die Erfassungszeit unabhängig von der Anzahl der Objekte im Verzeichnis niedrig bleibt.

Aktivieren von Dateiereignissen

Dateiereignisse erfordern eine einmalige Gewährung von Cloudberechtigungen sowie einen externen Speicherort, der für die Nutzung des Dienstes für verwaltete Dateiereignisse konfiguriert ist. Nach der Einrichtung können alle Autoloader-Streams, die von diesem externen Speicherort lesen, Dateiereignisse ohne zusätzliche Konfiguration verwenden.

1. Erteilen Sie den erforderlichen Cloudberechtigungen auf der Cloudanbieterseite. Die Anforderungen variieren je nach Cloudanbieter. Siehe Einrichten von Dateiereignissen für einen externen Speicherort.

2. Setzen Sie cloudFiles.useManagedFileEvents in Ihrer Auto-Loader-Abfrage auf true fest.

   df = (spark.readStream
     .format("cloudFiles")
     .option("cloudFiles.format", "json")
     .option("cloudFiles.useManagedFileEvents", "true")
     .load("/path/to/data/dir"))
   

   Die vollständigen Einrichtungsschritte finden Sie unter Zu Auto Loader mit Dateiereignissen migrieren.

Wenn Sie keine Dateiereignisse verwenden können

Möglicherweise können Sie Dateiereignisse nicht verwenden, wenn:

• Der externe Speicherort ist nicht für Dateiereignisse konfiguriert.
• Sicherheitsrichtlinien der Organisation lassen das Aktivieren von Dateiereignissen an einem freigegebenen externen Speicherort nicht zu.

Verwenden Sie in diesen Fällen den klassischen Dateibenachrichtigungsmodus oder den Verzeichnisauflistungsmodus. Einen vollständigen Vergleich der Dateierkennungsmodi finden Sie unter "Vergleichen der Dateierkennungsmodi für das automatische Laden".

Verwalten der Schemaentwicklung

Auto Loader leitet das Schema automatisch ab, aber wie Sie die Schemaentwicklung konfigurieren, wirkt sich auf die Vollständigkeit und Pipelinestabilität der Daten aus. Verwenden Sie die folgende Tabelle, um eine Strategie auszuwählen.

Nachdem Sie eine Strategie ausgewählt haben, wenden Sie die folgenden Methoden an, um das Verhalten der Schemaentwicklung zu optimieren.

Verwenden von Schemahinweisen für bekannte Feldtypen

Verwenden Sie die cloudFiles.schemaHints Option, um Typen für Felder zu erzwingen, die Sie im Voraus kennen, und gleichzeitig die Schemaeinleitung für andere Felder zulassen.

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaHints", "id long, amount double")
  .load("/path/to/data/dir"))

Verwenden Sie die Typverbreiterung für kompatible Typänderungen

Der addNewColumnsWithTypeWidening Schemaentwicklungsmodus erweitert automatisch kompatible Typen (z int . B. auf long) anstatt Daten an die _rescued_data Spalte weiterzuleiten. Dies vermeidet die Notwendigkeit von Nachbearbeitungsaufträgen zur Verarbeitung einfacher Typumwandlungen.

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "parquet")
  .option("cloudFiles.schemaEvolutionMode", "addNewColumnsWithTypeWidening")
  .load("/path/to/data/dir"))

Erfassung als Typ Variant für unvorhersehbare Schemata

Wenn Ihre Daten keinem bestimmten Schema entsprechen oder sich das Schema kontinuierlich ändert, nehmen Sie die Daten als Variant Typ ein.

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("singleVariantColumn", "data")
  .load("/path/to/data/dir"))

Variant ermöglicht Schema-on-Read zur Abfragezeit, ist jedoch weniger effizient als das Abfragen strukturierter Spalten. Die vollständige Mechanik der Schemaausleitung und -entwicklung finden Sie unter "Konfigurieren der Schemaausleitung und -evolution in Auto Loader".

Mit fehlerhaften Daten und Datenqualität umgehen

Die folgenden Methoden helfen Ihnen, fehlerhafte Daten zu erkennen, zu erfassen und zu isolieren, bevor sie auf nachgelagerte Ebenen verteilt wird.

Aktivieren _rescued_data und _corrupt_record

Auto Loader stellt zwei Spalten zur Erfassung von Daten bereit, die sich nicht fehlerfrei parsen lassen.

• _rescued_data erfasst Felder, die nicht mit dem aktuellen Schema übereinstimmen. Sie wird automatisch vom Auto Loader hinzugefügt.
• _corrupt_record erfasst Zeilen, die überhaupt nicht analysiert werden können. Aktivieren Sie sie mithilfe von columnNameOfCorruptRecord:

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaHints", "_corrupt_record string")
  .option("columnNameOfCorruptRecord", "_corrupt_record")
  .load("/path/to/data/dir"))

Databricks empfiehlt columnNameOfCorruptRecord gegenüber badRecordsPath, um mögliche Racebedingungen zu vermeiden, durch die beschädigte Datensätze übersehen werden können.

Verwenden Sie die Erwartungen von Lakeflow-Pipelines zur Überwachung

Legen Sie Erwartungen für Lakeflow-Pipelines fest, um zu überprüfen, ob _rescued_data und _corrupt_record unter normalen Bedingungen NULL sind. Werte ungleich NULL signalisieren Schemaabweichungen oder Datenbeschädigung.

import dlt

@dlt.table
@dlt.expect("no rescued data", "_rescued_data IS NULL")
@dlt.expect("no corrupt records", "_corrupt_record IS NULL")
def bronze_table():
    return (spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.schemaHints", "_corrupt_record string")
        .option("columnNameOfCorruptRecord", "_corrupt_record")
        .load("/path/to/data/dir"))

Isolieren beschädigter Daten

Isolieren Sie Zeilen, die nicht analysierbare Daten in einer dedizierten Senke enthalten, für die Untersuchung. Dadurch wird verhindert, dass beschädigte Daten auf nachgelagerte Ebenen verteilt werden.

import dlt

@dlt.table
def corrupt_records_sink():
    return dlt.read_stream("bronze_table").where("_corrupt_record IS NOT NULL")

@dlt.view
def clean_table():
    return dlt.read_stream("bronze_table").where("_corrupt_record IS NULL")

Kommentieren von Daten mit Quelldateimetadaten

Fügen Sie die Spalte _metadata in Ihre Auto-Loader-Ingestionsabfragen ein. Erfassen Sie mindestens file_path und file_modification_time. Auf diese Weise können Sie Datenprobleme auf bestimmte Quelldateien zurückverfolgen und sie mit cloud_files_state() verknüpfen, um den vollständigen Dateilebenszyklus nachzuvollziehen.

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .load("/path/to/data/dir")
  .select("*", "_metadata.file_path", "_metadata.file_modification_time"))

Ausführliche Informationen finden Sie in der Spalte "Dateimetadaten".

Optimieren von Kosten und Leistung

Die folgenden Methoden reduzieren die drei wichtigsten Kostentreiber für das automatische Laden: Cloud-API-Aufrufe LIST , Leerlaufberechnung und langfristiges Speicherwachstum.

• Verwenden von Dateiereignissen zum Minimieren LIST API-Kosten: Dateiereignisse stellen eine inkrementelle Dateiermittlung bereit, ohne dass für jede Ausführung vollständige Verzeichnisauflistungen erforderlich sind. Dies ist die einzige wirkungsvollste Kostenoptimierung für das automatische Laden.

• Verwenden Sie Dateiankunftstrigger für die ereignisgesteuerte Verarbeitung: Dateiankunftstrigger starten Ihre Pipeline nur, wenn neue Dateien ankommen, sodass Sie nicht für die Berechnung im Leerlauf bezahlen. Weitere Informationen finden Sie unter Auslösen von Aufträgen beim Eintreffen neuer Dateien.

• Verarbeitete Dateien mit cloudFiles.cleanSource archivieren: Verwenden Sie cloudFiles.cleanSource, um verarbeitete Dateien automatisch zu löschen oder zu verschieben. Dadurch werden sowohl Speicherkosten als auch Verzeichnisauflistungskosten für langlebige Datenströme reduziert. Ausführliche Informationen finden Sie unter "Archivierungsdateien im Quellverzeichnis", um die Kosten zu senken.

  • Verwenden Sie den Modus delete, um Dateien nach der Ingestion zu entfernen.
  • Verwenden Sie move den Modus zum Archivieren von Dateien an einem anderen Speicherort für Compliance oder Überwachung.

  df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.cleanSource", "delete")
    .load("/path/to/data/dir"))
  

  Warnung

  Aktivieren Sie cloudFiles.cleanSource nicht, wenn mehrere Auto-Loader-Streams oder andere Clients aus demselben Quellverzeichnis lesen.

• Nutzen Sie Leistungsverbesserungen: Aktualisieren Sie auf die neueste Databricks-Runtime oder verwenden Sie serverlose Compute, um von den letzten Leistungsverbesserungen des automatischen Ladeprogramms zu profitieren.

Prüfpunktverwaltung

Der Checkpoint speichert den Fortschritt des Datenstroms und den Dateistatus. Das Fehlkonfigurieren oder Verlieren des Prüfpunkts erfordert einen vollständigen Neustart. Behandeln Sie ihn also als kritische Infrastruktur.

• Wenden Sie niemals Cloudobjektlebenszyklusrichtlinien auf Prüfpunktstandorte an. Wenn Prüfpunktdateien gelöscht werden, ist der Datenstromzustand beschädigt, und Sie müssen von Grund auf neu starten.
• Verwenden Sie separate Prüfpunkte für jeden Datenstrom und jedes Quellverzeichnis.
• Erwägen Sie cloudFiles.maxFileAge für langlebige Datenströme mit hohem Volumen, um das Anwachsen des Zustands zu begrenzen. Verwenden Sie eine konservative Einstellung (mindestens 90 Tage empfohlen). Wenn Sie diesen Wert zu aggressiv festlegen, besteht die Gefahr, dass Dateien neu verarbeitet werden, die das automatische Ladeprogramm bereits aufgenommen hat, wenn sie außerhalb des Fensters liegen.

Ausführliche Informationen finden Sie unter Dateiereignisverfolgung.

Verwenden Sie Volumes für eine optimale Dateisuche mit Dateiereignissen

Erstellen Sie für mehr Leistung bei Dateiereignissen für jeden Pfad bzw. jedes Unterverzeichnis, aus dem Autoloader liest, ein externes Volume. Geben Sie für Autoloader Volumepfade (z. B. /Volumes/catalog/schema/volume) anstelle von Cloudpfaden (z. B. s3://bucket/path) an. Dies optimiert die Dateisuche durch ein optimiertes Datenzugriffsmuster.

Weitere Best Practices zu Datei-Ereignissen finden Sie unter Best Practices für Auto Loader mit Datei-Ereignissen.