Utwórz potok za pomocą potoków Lakeflow

Potoki danych produkcyjnych wymagają niezawodności, łatwości utrzymania i jasnego egzekwowania jakości danych. Jako inżynier danych prawdopodobnie poświęcasz dużo czasu na pisanie kodu w celu obsługi przetwarzania przyrostowego, organizowania zależności i weryfikowania jakości danych. Lakeflow pipelines w Azure Databricks pomagają sprostać tym wyzwaniom, umożliwiając zdefiniowanie, jak mają wyglądać dane, a nie jak przetwarzać je krok po kroku.

W tej lekcji dowiesz się, jak tworzyć potoki danych przy użyciu podejścia deklaratywnego, definiować tabele strumieniujące i zmaterializowane widoki oraz stosować oczekiwania dotyczące jakości danych w celu wymuszania ograniczeń na dane.

Omówienie podejścia deklaratywnego

Tradycyjne potoki danych wymagają pisania kodu imperatywnego, który określa każdy krok przetwarzania. Obsługujesz logikę przetwarzania przyrostowego, zarządzasz odzyskiwaniem punktu kontrolnego i organizujesz zależności między tabelami. W przypadku potoków Lakeflow zamiast tego deklarujesz pożądany stan końcowy, a struktura obsługuje szczegóły wykonywania.

Podejście deklaratywne zapewnia trzy kluczowe korzyści dla potoków produkcyjnych:

• Automatyczna aranżacja: struktura analizuje zależności między tabelami i określa poprawną kolejność wykonywania. Jeśli zdefiniujesz srebrną tabelę, z której odczytuje dane z brązowej tabeli, framework automatycznie przetwarza najpierw brązową tabelę.

• Przetwarzanie przyrostowe: tabele przesyłania strumieniowego przetwarzają każdy rekord dokładnie raz. Zmaterializowane widoki automatycznie identyfikują i przetwarzają tylko zmienione dane, gdy to możliwe, unikając kosztownych pełnych ponownych kompilacji.

• Wbudowana logika ponawiania prób: gdy wystąpią błędy, framework ponawia próbę na najbardziej szczegółowym poziomie — najpierw na poziomie zadania Spark, a następnie na poziomie przepływu, a na koniec na poziomie potoku.

Rozważmy scenariusz, w którym tworzysz pipeline analityczny dla transakcji klientów. Zamiast pisać setki wierszy kodu Structured Streaming z zarządzaniem punktami kontrolnymi i obsługą stanu, deklarujesz swoje tabele i pozwalasz frameworkowi zarządzać złożonością.

Definiowanie tabel strumieniowych na potrzeby załadunku danych

Tabela strumieniowa to tabela delty zoptymalizowana pod kątem przetwarzania danych dołączanych. Każdy rekord wejściowy jest przetwarzany dokładnie raz, dzięki czemu tabele przesyłania strumieniowego idealnie nadają się do pozyskiwania danych ze źródeł, takich jak magazyn w chmurze lub kolejki komunikatów.

Poniższy przykład SQL tworzy tabelę strumieniową, która pozyskuje pliki JSON z magazynu w chmurze przy użyciu Auto Loader.

CREATE OR REFRESH STREAMING TABLE customers_bronze
AS SELECT * FROM STREAM read_files(
  "/Volumes/raw_data/customers",
  format => "json"
);

Słowo kluczowe STREAM informuje potok o traktowaniu źródła jako zestawu danych przesyłania strumieniowego. Przy każdej aktualizacji potoku przetwarzane i dołączane do tabeli są tylko nowe pliki.

Ten sam wynik można osiągnąć za pomocą języka Python:

from pyspark import pipelines as dp

@dp.table
def customers_bronze():
  return (
    spark.readStream.format("cloudFiles")
     .option("cloudFiles.format", "json")
     .option("cloudFiles.inferColumnTypes", "true")
     .load("/Volumes/raw_data/customers")
  )

Dekorator @dp.table oznacza tę funkcję jako strumieniową definicję tabeli. Funkcja zwraca strumieniowy DataFrame, który framework przetwarza stopniowo przy każdej aktualizacji.

Tworzenie zmaterializowanych widoków dla przekształceń

Podczas gdy tabele przesyłania strumieniowego obsługują pozyskiwanie wyłącznie typu append-only, zmaterializowane widoki są lepiej dostosowane do przekształceń, w których konieczne jest wnioskowanie o agregacjach lub wymiarach, które mogą się zmieniać. Zmaterializowany widok buforuje wyniki zapytania i automatycznie synchronizuje je ze zmianami nadrzędnymi.

Poniższy przykład tworzy zmaterializowany widok, który łączy dane klienta i transakcji:

CREATE OR REPLACE MATERIALIZED VIEW customer_transactions
AS SELECT 
  c.customer_id,
  c.customer_name,
  c.region,
  t.transaction_date,
  t.amount
FROM customers c
INNER JOIN transactions t ON c.customer_id = t.customer_id;

W przeciwieństwie do tabel przesyłania strumieniowego zmaterializowane widoki obsługują złożone zapytania, w tym agregacje i sprzężenia. Gdy tabele nadrzędne zmieniają się, framework określa najbardziej wydajny sposób aktualizowania zmaterializowanego widoku — często przetwarzając tylko dane, których to dotyczy, zamiast ponownego przeliczania wszystkiego.

W języku Python użyjesz dekoratora @dp.materialized_view :

from pyspark import pipelines as dp

@dp.materialized_view
def regional_sales():
  customers_df = spark.read.table("customers")
  transactions_df = spark.read.table("transactions")
  
  return (
    customers_df.join(transactions_df, on="customer_id", how="inner")
    .groupBy("region")
    .agg({"amount": "sum"})
  )

Stosowanie oczekiwań dotyczących jakości danych

Potoki produkcyjne wymagają zapewniania jakości danych. Oczekiwania to ograniczenia deklaratywne, które weryfikują rekordy w trakcie przepływu przez potok. Definiujesz, jak wyglądają prawidłowe dane, a platforma śledzi metryki i podejmuje działania w przypadku wystąpienia naruszeń.

W przypadku niepowodzenia walidacji rekordu platforma może reagować na różne sposoby w zależności od określonej akcji:

Użyj ON VIOLATION FAIL UPDATE w przypadku krytycznych ograniczeń, w których każde naruszenie wskazuje poważny problem z danymi, który wymaga zbadania przed kontynuowaniem przetwarzania.

Opracowywanie potoków za pomocą Edytora Potoków Lakeflow

Edytor potoków Lakeflow oferuje zintegrowane środowisko programistyczne do tworzenia i testowania potoków. Podczas tworzenia nowego potoku ETL w usłudze Azure Databricks edytor udostępnia domyślną strukturę folderów z oddzielnymi katalogami dla kodu źródłowego przekształcania, notesów eksploracyjnych i modułów narzędziowych.

Edytor obsługuje programowanie iteracyjne za pomocą kilku funkcji:

• Genie Code: Opisz potok w języku naturalnym i pozwól temu agentowi tworzyć, aktualizować i usuwać błędy w kodzie potoku — od odnajdywania danych i generowania kodu potoku aż po egzekucję potoku i rozwiązywanie problemów z jakością danych.

• Przebieg suchy: weryfikuje kod potoku bez przetwarzania danych, co pozwala przechwytywać błędy składniowe i brakujące zależności przed wykonaniem.

• Selektywne wykonywanie: uruchamianie pojedynczych plików lub definicji pojedynczej tabeli, a nie całego potoku, co umożliwia szybszą iterację podczas programowania.

• Interaktywny DAG: wizualizacja wykresu zależności między tabelami, wybór wielu tabel do selektywnego odświeżania i sprawdzanie metryk wykonania.

• Podgląd danych: przykładowe dane z tabel strumieniowych i zmaterializowanych widoków bezpośrednio w edytorze do weryfikacji logiki przekształcania.

Dzięki podejściu deklaratywnemu i zintegrowanemu narzędziu można tworzyć potoki danych klasy produkcyjnej, które są obsługiwane, zauważalne i wymuszają spójną jakość danych. Struktura obsługuje złożoność przetwarzania przyrostowego i orkiestracji, dzięki czemu można skupić się na zdefiniowaniu logiki biznesowej, która ma znaczenie dla organizacji.