Najlepsze rozwiązania dotyczące automatycznego modułu ładującego

Na tej stronie opisano najlepsze praktyki, które można zastosować podczas konfigurowania narzędzia Auto Loader, aby działało niezawodnie, ekonomicznie i na dużą skalę w Twoim przypadku użycia.

Te sprawdzone praktyki zmniejszają obciążenie operacyjne i pomagają zapobiegać typowym problemom, które trudno zdiagnozować w środowisku produkcyjnym, takim jak: niepotrzebne koszty LIST API wynikające z pełnego skanowania katalogów, cicha utrata danych spowodowana zmianami schematu oraz ponowne uruchomienia potoku spowodowane błędną konfiguracją checkpointu.

Aby uzyskać szczegółowe informacje o konfiguracji produkcyjnej, zobacz Konfigurowanie automatycznego modułu ładującego dla obciążeń produkcyjnych. Aby uzyskać informacje na temat monitorowania i obserwacji, zobacz Monitorowanie i obserwowanie automatycznego modułu ładującego.

Wybierz odpowiednie środowisko uruchomieniowe

Najlepsza platforma wykonawcza dla Twojego przypadku użycia zależy od tego, jak dużej kontroli nad potokiem potrzebujesz i jak duże obciążenie operacyjne chcesz przejąć na siebie. Dla większości użytkowników i potoków produkcyjnych Auto Loader z potokami Lakeflow dobrze się sprawdza. Jeśli jednak potrzebujesz maksymalnej kontroli i możliwości dostosowywania, użyj funkcji Auto Loader ze Structured Streaming. Aby uzyskać najprostszą konfigurację z funkcją zarządzaną, użyj zarządzanego łącznika LakeFlow, jeśli jest dostępny.

Potoki Lakeflow rozszerzają Structured Streaming o automatyczne skalowanie, kontrole jakości danych, obsługę ewolucji schematu i monitorowanie za pomocą dziennika zdarzeń. Databricks zaleca potoki Lakeflow do większości produkcyjnych zadań związanych z pozyskiwaniem danych.

Wybieranie odpowiedniego typu planowania i wyzwalacza

Najlepszy typ harmonogramu i wyzwalacza dla danego zastosowania zależy od wymagań dotyczących opóźnień oraz sposobu napływu plików. W większości zastosowań Databricks zaleca wyzwalacz po pojawieniu się pliku z włączoną obsługą zdarzeń plików. Pozwala to na pozyskiwanie danych z niskimi opóźnieniami przy niskich kosztach, ponieważ zasoby obliczeniowe są uruchamiane tylko wtedy, gdy pojawiają się nowe pliki. Trzy typy wyzwalaczy różnią się w zależności od tego, kiedy i jak często uruchamia się potok:

• Ciągły: potok działa bez zatrzymywania. Używaj tego tylko wtedy, gdy opóźnienie poniżej jednej sekundy jest bezwzględnym wymogiem, ponieważ ciągłe przetwarzanie jest droższe. Powiąż ze zdarzeniami plikowymi.
• Wyzwalacz pojawienia się pliku: potok rozpoczyna się, gdy w lokalizacji źródłowej pojawią się nowe pliki. Najlepsze w przypadku małych do średnich opóźnień lub nieregularnych wzorców przybycia plików. Wymaga włączenia zdarzeń dotyczących plików. Zobacz Wyzwalanie zadań po nadejściu nowych plików.
• Zaplanowane: potok jest uruchamiany zgodnie z harmonogramem opartym na czasie (na przykład co godzinę). Użyj, gdy wymagania dotyczące opóźnienia są łagodne (od minut do godzin). Działa z listowaniem katalogów, ale zdarzenia plików zmniejszają koszty nawet w trybie harmonogramu, ponieważ pozwalają uniknąć pełnego skanowania katalogów.

Szczegółowe informacje na temat używania elementu Trigger.AvailableNow do planowania wsadowego można znaleźć w artykule Używanie Trigger.AvailableNow i ograniczanie szybkości.

Wybieranie odpowiedniego trybu odnajdywania plików

Moduł automatycznego ładowania obsługuje trzy tryby odnajdywania plików z różnymi kompromisami w zakresie złożoności konfiguracji, skalowalności i kosztów.

Zdarzenia plików konsolidują zasoby pamięci masowej w chmurze, używając jednej subskrypcji i jednej kolejki na każdą lokalizację zewnętrzną zamiast jednej subskrypcji i jednej kolejki na każdy strumień. Różnica w wydajności jest znacząca na dużą skalę: lista katalogów musi skanować cały katalog źródłowy na każdym wyzwalaczu, więc czas pozyskiwania rośnie wraz z rozmiarem katalogu. Zdarzenia dotyczące plików przekazują powiadomienia o nowych plikach bezpośrednio, więc czas przetwarzania pozostaje niski bez względu na liczbę obiektów w katalogu.

Włącz zdarzenia plików

Zdarzenia plików wymagają jednorazowego udzielenia uprawnień w chmurze i lokalizacji zewnętrznej skonfigurowanej do korzystania z zarządzanej usługi zdarzeń plików. Po skonfigurowaniu wszystkie strumienie Auto Loader odczytujące dane z tej zewnętrznej lokalizacji mogą używać zdarzeń plików bez dodatkowej konfiguracji.

1. Udziel wymaganych uprawnień do chmury po stronie dostawcy usług w chmurze. Wymagania różnią się w zależności od dostawcy usług w chmurze. Zobacz Skonfiguruj zdarzenia plików dla lokalizacji zewnętrznej.

2. Ustaw cloudFiles.useManagedFileEvents na true w zapytaniu Auto Loader.

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

   Pełne instrukcje konfiguracji znajdują się w sekcji Migracja do Auto Loader z użyciem zdarzeń plików.

Gdy nie można używać zdarzeń związanych z plikami

Korzystanie ze zdarzeń plików może nie być możliwe, gdy:

• Lokalizacja zewnętrzna nie jest skonfigurowana do obsługi zdarzeń plików.
• Zasady zabezpieczeń organizacji nie pozwalają na włączenie zdarzeń dotyczących plików we współdzielonej lokalizacji zewnętrznej.

W takich przypadkach należy użyć klasycznego trybu powiadamiania o plikach lub trybu wyświetlania listy katalogów. Aby zapoznać się z pełnym porównaniem trybów wykrywania plików, zobacz Porównanie trybów wykrywania plików automatycznego modułu ładującego.

Zarządzanie ewolucją schematu

Auto Loader automatycznie wywnioskuje schemat, ale sposób konfiguracji ewolucji schematu wpływa na kompletność danych i stabilność potoku danych. Skorzystaj z poniższej tabeli, aby wybrać strategię.

Po wybraniu strategii zastosuj następujące praktyki, aby precyzyjnie określić sposób działania ewolucji schematu.

Używanie wskazówek schematu dla znanych typów pól

cloudFiles.schemaHints Użyj opcji , aby wymusić typy dla pól, które znasz wcześniej, a jednocześnie zezwalaj na wnioskowanie schematu dla innych pól.

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

Użyj rozszerzania typu dla zgodnych zmian typu

addNewColumnsWithTypeWidening Tryb ewolucji schematu automatycznie rozszerza zgodne typy (na przykład int do long) zamiast kierowania danych do kolumny_rescued_data. Pozwala to uniknąć konieczności wykonywania zadań przetwarzania końcowego w celu obsługi prostych promocji typów.

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

Importowanie jako typ Variant w przypadku nieprzewidywalnych schematów

Jeśli dane nie są zgodne z żadnym konkretnym schematem lub schemat stale się zmienia, zaimportuj dane jako typ Variant.

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

Variant zapewnia odczyt schematu w czasie zapytania, ale jest mniej wydajny niż wykonywanie zapytań dotyczących kolumn strukturalnych. Aby uzyskać pełną mechanikę wnioskowania i ewolucji schematu, zobacz Konfigurowanie wnioskowania schematu i ewolucji w module automatycznego ładowania.

Obsługiwać nieprawidłowe dane i zarządzać jakością danych

Poniższe rozwiązania ułatwiają wykrywanie, przechwytywanie i izolowanie nieprawidłowych danych przed ich propagowaniem do warstw podrzędnych.

Włącz _rescued_data i _corrupt_record

Auto Loader oferuje dwie kolumny do wychwytywania danych, których nie można poprawnie przeanalizować składniowo.

• _rescued_data przechwytuje pola, które nie są zgodne z bieżącym schematem. Jest dodawany automatycznie przez Auto Loader.
• _corrupt_record przechwytuje wiersze, których nie można przeanalizować w ogóle. Włącz go przy użyciu polecenia 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 zaleca użycie columnNameOfCorruptRecord zamiast badRecordsPath, aby uniknąć potencjalnych stanów wyścigu, przez które można pominąć uszkodzone rekordy.

Korzystanie z oczekiwań potoków lakeflow na potrzeby monitorowania

Skonfiguruj oczekiwania dla potoków Lakeflow, aby zweryfikować, czy _rescued_data i _corrupt_record są NULL w normalnych warunkach. Dryf schematu sygnału o wartościach innych niż NULL lub uszkodzenie danych.

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

Izolowanie uszkodzonych danych

Izoluj wiersze zawierające dane, których nie można przeanalizować, w dedykowanym ujściu docelowym do zbadania. Zapobiega to propagowaniu uszkodzonych danych do warstw podrzędnych.

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

Dodawanie adnotacji do danych za pomocą metadanych pliku źródłowego

Uwzględnij kolumnę _metadata w zapytaniach ładowania danych Auto Loader. Przynajmniej uwzględnij file_path i file_modification_time. Dzięki temu można śledzić problemy z danymi do określonych plików źródłowych i łączyć je z danymi z cloud_files_state() dla całego cyklu życia pliku.

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

Aby uzyskać szczegółowe informacje, zobacz Kolumna metadanych pliku.

Optymalizowanie kosztów i wydajności

Poniższe praktyki ograniczają trzy główne źródła kosztów związanych z funkcją Auto Loader: wywołania interfejsu API chmury LIST, bezczynne zasoby obliczeniowe oraz długoterminowy przyrost wykorzystania pamięci masowej.

• Użyj zdarzeń plików, aby zminimalizować LIST Koszty interfejsu API: Zdarzenia plików zapewniają przyrostowe odnajdywanie plików, eliminując konieczność tworzenia pełnych list katalogów w każdym uruchomieniu. Jest to pojedyncza najbardziej wpływowa optymalizacja kosztów dla automatycznego modułu ładującego.

• Użyj wyzwalaczy przybycia plików na potrzeby przetwarzania sterowanego zdarzeniami: wyzwalacze przybycia plików uruchamiają potok tylko po nadejściu nowych plików, więc nie płacisz za bezczynne obliczenia. Zobacz Wyzwalanie zadań po nadejściu nowych plików.

• Archiwizowanie przetworzonych plików za pomocą pliku cloudFiles.cleanSource: służy cloudFiles.cleanSource do automatycznego usuwania lub przenoszenia przetworzonych plików. Zmniejsza to zarówno koszty przechowywania, jak i koszty wyświetlania zawartości katalogów w przypadku długotrwałych strumieni. Aby uzyskać szczegółowe informacje, zobacz Archiwizowanie plików w katalogu źródłowym, aby obniżyć koszty.

  • Użyj trybu delete, aby usunąć pliki po zaimportowaniu.
  • Użyj move trybu, aby zarchiwizować pliki w innej lokalizacji pod kątem zgodności lub inspekcji.

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

  Warning

  Nie należy włączać cloudFiles.cleanSource, jeśli wiele strumieni Auto Loader lub innych klientów odczytuje dane z tego samego katalogu źródłowego.

• Skorzystaj z ulepszeń wydajności: uaktualnij do najnowszego środowiska Databricks Runtime lub użyj bezserwerowego środowiska obliczeniowego, aby skorzystać z najnowszych ulepszeń wydajności modułu automatycznego ładowania.

Zarządzanie punktami kontrolnymi

Punkt kontrolny przechowuje postęp strumienia i stan pliku. Błędna konfiguracja lub utrata punktu kontrolnego wymaga pełnego ponownego uruchomienia, dlatego traktuj ją jako infrastrukturę krytyczną.

• Przechowuj punkt kontrolny wraz z wyjściem strumienia, a nie z jego danymi źródłowymi. Punkt kontrolny przechowuje stan zapytania streamingowego, więc trzymaj je w trwałej pamięci po stronie wyjściowej, takiej jak dedykowany wolumin katalogu Unity, a nie w katalogu źródłowym skanowanym przez Auto Loader.
• Nigdy nie stosuj zasad cyklu życia obiektów w chmurze do lokalizacji punktów kontrolnych. Jeśli pliki punktów kontrolnych są usuwane, stan strumienia jest uszkodzony i musisz ponownie uruchomić od podstaw.
• Użyj oddzielnych punktów kontrolnych dla każdego strumienia i katalogu źródłowego.
• Rozważ użycie cloudFiles.maxFileAge w przypadku długotrwałych strumieni o dużym wolumenie, aby ograniczyć przyrost stanu. Użyj ustawienia konserwatywnego (zalecane minimum 90 dni). Ustawienie tej wartości zbyt agresywnie wiąże się z ryzykiem ponownego przetwarzania plików, które moduł automatycznego ładowania już pozyskał, jeśli spadną poza okno.

Aby uzyskać szczegółowe informacje, zobacz Śledzenie zdarzeń plików.

Najlepsze praktyki dotyczące zdarzeń plików

Najlepsze praktyki dotyczące zdarzeń plików znajdziesz w artykule Najlepsze praktyki dla Auto Loader z zdarzeniami plików.