Implementowanie kontroli typów danych
Wskazówka
Aby uzyskać więcej szczegółów, zobacz kartę Tekst i obrazy .
Każda kolumna w zestawie danych ma oczekiwany typ danych, a gdy dane przychodzące nie spełniają tych oczekiwań, problemy spływają w dół. Wartość typu ciąg znaków w miejscu, gdzie powinna być liczba całkowita, może zakłócać obliczenia, powodować błędy agregacji lub awarie pipeline'u. Sprawdzanie typu danych zapewnia, że każde pole zawiera wartości poprawnego typu przed ich wprowadzeniem do tabel.
W tej lekcji dowiesz się, jak zaimplementować kontrole typów danych przy użyciu wymuszania schematu, jawnego rzutowania typów i ograniczeń weryfikacji w usłudze Azure Databricks.
Zrozumienie wymuszania schematu
Usługa Delta Lake automatycznie weryfikuje typy danych podczas zapisywania danych w tabeli. Ten wbudowany mechanizm, nazywany wymuszaniem schematu, porównuje typy danych kolumn przychodzących ze schematem tabeli docelowej. Gdy wystąpi niezgodność typów, usługa Delta Lake próbuje bezpiecznie rzutować wartość. Jeśli rzutowanie zakończy się niepowodzeniem, operacja zapisu zgłosi błąd.
Wymuszanie schematu podczas operacji wstawiania stosuje następujące reguły:
- Wszystkie wstawione kolumny muszą istnieć w tabeli docelowej
- Wszystkie typy danych w kolumnach muszą być zgodne lub można je bezpiecznie rzutować na typy docelowe.
Rozważ tabelę z kolumną całkowitą:
CREATE TABLE inventory (
product_id INT,
quantity INT,
last_updated DATE
);
W przypadku wstawiania danych, gdzie quantity jest ciągiem reprezentującym liczbę, Delta Lake rzutuje go na typ docelowy.
-- This succeeds because '100' can be cast to INT
INSERT INTO inventory VALUES (1, '100', '2026-01-15');
Jednak wstawianie ciągu, którego nie można przekonwertować na liczbę całkowitą, powoduje niepowodzenie operacji:
-- This fails because 'fifty' cannot be cast to INT
INSERT INTO inventory VALUES (2, 'fifty', '2026-01-15');
Egzekucja schematu zapewnia pierwszą linię obrony przed niezgodnościami typów danych. Aby uzyskać większą kontrolę nad sposobem obsługi niezgodności, można użyć jawnego rzutowania.
Stosuj rzutowanie jawne typu
Funkcja cast konwertuje wartości z jednego typu danych na inny. Gdy konwersja nie powiedzie się, zostanie zgłoszony błąd. Funkcja try_cast działa w taki sam sposób, ale zwraca NULL zamiast zgłaszać błąd w przypadku niepowodzenia konwersji.
Użyj cast, jeśli chcesz uzyskać ścisłą walidację, która zatrzymuje przetwarzanie w przypadku nieprawidłowych danych.
SELECT
cast(raw_amount AS DECIMAL(10,2)) AS amount,
cast(raw_date AS DATE) AS transaction_date
FROM staging_data;
Użyj try_cast, aby zidentyfikować nieprawidłowe wartości, nie powodując niepowodzenia zapytania.
SELECT
raw_amount,
try_cast(raw_amount AS DECIMAL(10,2)) AS validated_amount,
CASE
WHEN try_cast(raw_amount AS DECIMAL(10,2)) IS NULL
THEN 'Invalid amount format'
ELSE 'Valid'
END AS validation_status
FROM staging_data;
W przypadku try_cast rekordy z nieprawidłowymi wartościami zwracają kolumnę NULL dla rzutu. Następnie można filtrować, flagować lub poddać kwarantannie te rekordy na podstawie wymagań dotyczących jakości danych.
Implementowanie walidacji typu z ograniczeniami
Ograniczenia CHECK umożliwiają definiowanie niestandardowych reguł walidacji, które są wymuszane za każdym razem, gdy dane zostaną wstawione lub zaktualizowane. Zazwyczaj używane w przypadku zakresów wartości i wzorców można łączyć je z funkcjami obsługującymi typy, aby tworzyć zaawansowane kontrole typów.
Można na przykład sprawdzić, czy kolumna ciągu zawiera tylko znaki liczbowe:
CREATE TABLE orders (
order_id INT,
order_total STRING
);
ALTER TABLE orders
ADD CONSTRAINT valid_order_total CHECK (order_total REGEXP '^[0-9]+(\\.[0-9]+)?$')
To ograniczenie gwarantuje, że kolumna order_total zawiera wartości, które wyglądają jak prawidłowe liczby, przechwytując źle sformułowane dane przed wejściem do tabeli.
Na potrzeby weryfikacji daty można użyć try_cast funkcji w ramach ograniczenia:
ALTER TABLE events
ADD CONSTRAINT valid_event_date CHECK (try_cast(event_date_str AS DATE) IS NOT NULL);
Takie podejście odrzuca rekordy, w których event_date_str nie ma prawidłowego formatu daty.
Ważne
Przed dodaniem ograniczenia do istniejącej tabeli usługa Azure Databricks sprawdza, czy wszystkie istniejące wiersze spełniają ograniczenie. Zaplanuj ten krok weryfikacji, szczególnie w przypadku dużych tabel.
Zarządzanie niezgodnościami typów w potokach
Podczas przetwarzania danych ze źródeł zewnętrznych typowe są niezgodności typów. Zaimplementuj wzorzec, który oddziela prawidłowe rekordy od nieprawidłowych rekordów, umożliwiając przetwarzanie dobrych danych podczas kwarantowania problematycznych rekordów do przeglądu:
-- Insert valid records into the target table
INSERT INTO silver_transactions
SELECT
transaction_id,
cast(amount AS DECIMAL(10,2)) AS amount,
cast(transaction_date AS DATE) AS transaction_date
FROM bronze_transactions
WHERE try_cast(amount AS DECIMAL(10,2)) IS NOT NULL
AND try_cast(transaction_date AS DATE) IS NOT NULL;
-- Capture invalid records for investigation
INSERT INTO quarantine_transactions
SELECT
transaction_id,
amount AS raw_amount,
transaction_date AS raw_date,
current_timestamp() AS quarantined_at,
'Type validation failed' AS reason
FROM bronze_transactions
WHERE try_cast(amount AS DECIMAL(10,2)) IS NULL
OR try_cast(transaction_date AS DATE) IS NULL;
Ten schemat gwarantuje, że przepływ pracy będzie nadal przetwarzać prawidłowe dane, zachowując nieprawidłowe rekordy na potrzeby późniejszej analizy.
Używanie parametrów przetwarzania potokowego do sprawdzania typów
Potoki Lakeflow zapewniają expectations, które umożliwiają definiowanie reguł jakości danych bezpośrednio w definicjach potoku. Możesz użyć oczekiwań, aby sprawdzić, czy wartości można rzutować do oczekiwanych typów:
from pyspark import pipelines as dp
@dp.table
@dp.expect_or_drop("valid_amount", "try_cast(amount AS DECIMAL(10,2)) IS NOT NULL")
@dp.expect_or_drop("valid_date", "try_cast(event_date AS DATE) IS NOT NULL")
def validated_transactions():
return spark.readStream.table("raw_transactions")
Rekordy w expect_or_drop, które nie przechodzą sprawdzania typu, są porzucane przed dotarciem do tabeli docelowej. Użyj expect, aby rejestrować naruszenia bez porzucania rekordów, lub expect_or_fail do zatrzymywania potoku w przypadku wystąpienia naruszeń.
W przypadku potoków opartych na języku SQL:
CREATE OR REFRESH STREAMING TABLE validated_transactions (
CONSTRAINT valid_amount EXPECT (try_cast(amount AS DECIMAL(10,2)) IS NOT NULL) ON VIOLATION DROP ROW,
CONSTRAINT valid_date EXPECT (try_cast(event_date AS DATE) IS NOT NULL) ON VIOLATION DROP ROW
)
AS SELECT * FROM STREAM(raw_transactions);
Oczekiwania potoku integrują sprawdzanie poprawności typów danych bezpośrednio z logiką ETL, zapewniając widoczność metryk jakości danych poprzez interfejs użytkownika potoku.
Implementowanie kontroli typów danych na wielu poziomach — wymuszanie schematu, jawne rzutowanie, ograniczenia i oczekiwania potoku — tworzy ochronę w głębi systemu. Każda warstwa wychwytuje problemy, które mogą zostać przeoczone przez inne, co skutkuje wyższą jakością danych w tabelach Unity Catalog.