Pozyskiwanie danych za pomocą Auto Loader

Gdy nowe pliki danych są stale dostarczane w magazynie w chmurze, potrzebujesz wydajnego sposobu ich przetwarzania bez ręcznego śledzenia, które pliki zostały pozyskane. ** Auto Loader rozwiązuje to wyzwanie, automatycznie wykrywając i pozyskując nowe pliki, gdy tylko się pojawią, zapewniając gwarancje dokładnie jednokrotnego przetwarzania bez konieczności ręcznego zarządzania stanem lub punktami kontrolnymi.

Informacje o sposobie działania modułu ładującego automatycznego

Auto Loader to źródło strukturalnego przesyłania strumieniowego, które obserwuje lokalizację przestrzeni dyskowej w chmurze i jest używane do przetwarzania przyrostowego nowych plików. Zamiast wielokrotnie skanować całe katalogi, Auto Loader utrzymuje informację o przetworzonych już plikach, zapewniając, że każdy plik jest załadowany dokładnie raz.

Po uruchomieniu strumienia Auto Loader może on przetwarzać istniejące pliki w katalogu, a następnie stale obserwować nowe pliki. Strumień przechowuje informacje o postępie w lokalizacji punktu kontrolnego, co umożliwia wznowienie dokładnie z miejsca, w którym zostało zatrzymane, jeśli zostanie przerwane.

Moduł automatycznego ładowania wykrywa nowe pliki przy użyciu jednego z dwóch trybów:

• Tryb listy katalogów: Auto Loader okresowo wyświetla katalog wejściowy w celu odnalezienia nowych plików. Takie podejście nie wymaga dodatkowej konfiguracji poza dostępem do magazynu.
• Tryb powiadamiania o plikach: Funkcja automatycznego ładowania używa usług powiadomień w chmurze do odbierania zdarzeń po nadejściu nowych plików. Ten tryb jest bardziej wydajny w przypadku obciążeń na dużą skalę, ponieważ pozwala uniknąć powtarzających się list katalogów.

W przypadku większości obciążeń produkcyjnych Databricks zaleca tryb powiadamiania o plikach ze zdarzeniami plików włączonymi w zewnętrznej lokalizacji w Unity Catalog. W przypadku zdarzeń plików funkcja automatycznego ładowania odbiera powiadomienia bezpośrednio po odebraniu plików, co zmniejsza opóźnienia i koszty interfejsu API w chmurze.

Skonfiguruj Auto Loader do wczytywania plików

Aby załadować dane za pomocą Auto Loader, należy użyć formatu cloudFiles z spark.readStream. Poniższy przykład odczytuje pliki JSON z usługi Azure Data Lake Storage i zapisuje je w tabeli Unity Catalog.

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

Opcja cloudFiles.format określa format pliku źródłowego. Auto loader obsługuje pliki JSON, CSV, Parquet, Avro, ORC, XML, text i binarne. cloudFiles.schemaLocation przechowuje wywnioskowany schemat, który umożliwia modułowi ładującemu automatyczne śledzenie zmian schematu w czasie. Chociaż lokalizacja schematu i lokalizacja punktu kontrolnego mogą być takie same, przechowywanie ich oddzielnie umożliwia zresetowanie strumienia bez utraty schematu wnioskowanego.

Ustawienie trigger(availableNow=True) przetwarza wszystkie aktualnie dostępne pliki, a następnie zatrzymuje się. Ten schemat sprawdza się dobrze w przypadku zaplanowanych zadań wsadowych. W przypadku ciągłego przetwarzania można pominąć wyzwalacz lub użyć trigger(processingTime="1 minute") do przetwarzania plików w określonych odstępach czasu.

Używanie składni SQL z funkcją read_files

Auto Loader obsługuje również składnię SQL przez funkcję zwracającą tabelę read_files. Takie podejście jest przydatne, jeśli wolisz program SQL lub podczas pracy w notesach SQL:

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

W przypadku korzystania z read_files z tabel przesyłania strumieniowego w potokach usługi Lakeflow funkcje Auto Loader są automatycznie włączone. Użyj słowa kluczowego STREAM , aby włączyć przetwarzanie przyrostowe:

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

W tym kontekście pipeline automatycznie zarządza punktami kontrolnymi i ewolucją schematu.

Obsługa wnioskowania i ewolucji schematu

Moduł automatycznego ładowania może automatycznie wykrywać schemat plików źródłowych, eliminując konieczność ręcznego definiowania schematów. Po włączeniu wnioskowania schematu, Auto Loader pobiera próbki plików w celu określenia nazw kolumn i typów danych.

W przypadku plików JSON i CSV moduł automatycznego ładowania domyślnie wywnioskuje wszystkie kolumny jako ciągi. Takie konserwatywne podejście zapobiega niezgodnościom typów, które mogą spowodować utratę danych. Aby włączyć automatyczne wykrywanie typów, ustaw cloudFiles.inferColumnTypes opcję:

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

Gdy dane źródłowe wprowadzają nowe kolumny, moduł automatycznego ładowania wykrywa zmianę i może automatycznie rozwijać schemat docelowy. Opcja cloudFiles.schemaEvolutionMode steruje tym zachowaniem:

Uratowana kolumna danych (_rescued_data) przechwytuje wszystkie dane, które nie są zgodne z oczekiwanym schematem, w tym niezgodności typów i nieoczekiwanych kolumn. Ta funkcja zapobiega utracie danych, gdy dane źródłowe nie są zgodne z oczekiwaniami. Moduł automatycznego ładowania automatycznie dodaje tę kolumnę podczas wnioskowania schematu.

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

Aby zmienić nazwę uratowanych kolumn danych, użyj rescuedDataColumn opcji:

.option("rescuedDataColumn", "my_rescued_data")

Jeśli wiesz, że niektóre kolumny powinny mieć określone typy, użyj wskazówek schematu , aby zastąpić wywnioskowane typy:

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

Monitorowanie i optymalizacja strumieni Auto Loader

W przypadku obciążeń produkcyjnych monitorowanie ułatwia śledzenie postępu przetwarzania danych i identyfikowanie problemów. Funkcja automatycznego ładowania udostępnia funkcję SQL do wykonywania zapytań dotyczących stanu odnalezionych plików:

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

To zapytanie zwraca metadane dotyczące plików odnalezionych przez moduł automatycznego ładowania, w tym ich stan przetwarzania. Te informacje umożliwiają sprawdzenie, czy pliki są pozyskiwane zgodnie z oczekiwaniami.

Auto Loader przekazuje metryki za pomocą pulpitu nawigacyjnego postępu Structured Streaming, w tym:

• numFilesOutstanding: Pliki odnalezione, ale nie zostały jeszcze przetworzone
• numBytesOutstanding: Łączna liczba bajtów oczekujących na przetworzenie

W przypadku obciążeń o dużym wolumenie steruj szybkością przetwarzania przy użyciu cloudFiles.maxFilesPerTrigger lub cloudFiles.maxBytesPerTrigger. Te ustawienia zapobiegają sytuacji, w której jakakolwiek pojedyncza mikropartia staje się zbyt duża:

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

Aby uzyskać dostęp do metadanych na poziomie pliku w pozyskanych danych, wybierz kolumnę _metadata . Ta ukryta kolumna zawiera informacje o każdym pliku źródłowym, w tym file_path, file_name, file_sizei file_modification_time:

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

Moduł automatycznego ładowania bezproblemowo integruje się z potokami Lakeflow, gdzie automatycznie obsługuje zarządzanie punktami kontrolnymi i schematem. W następnej jednostce dowiesz się, jak używać Auto Loader w definicjach potoków deklaratywnych dla kompleksowych procesów pozyskiwania danych.