検証チェックを実装する

データ パイプラインでは、多くの場合、欠損値、重複する識別子、または許容範囲外の値を含むレコードが検出されます。 検証チェックを実施しないと、これらのデータ品質の問題がダウンストリームに伝達され、不適切な分析、失敗したレポート、信頼性の低いビジネス上の意思決定が発生します。 検証チェックは、データ インジェストの時点でこれらの問題をキャッチし、品質の高いデータのみがパイプラインを通過することを保証します。

このユニットでは、Lakeflowパイプラインのエクスペクテーションと Delta Lake のテーブル制約を使用して、NULL 許容性、データのカーディナリティ、範囲チェックの検証を実装する方法について説明します。

Azure Databricks の検証アプローチを理解する

Azure Databricks には、検証チェックを実装するための 2 つの主要なメカニズム (パイプラインの期待とテーブルの制約) が用意されています。 各アプローチは異なるシナリオに対応し、個別の機能を提供します。

パイプラインの期待値は、Lakeflow パイプラインのデータ変換中に検証を適用する機能です。 データがルールに違反した場合に、警告を出したり、無効なレコードを破棄したり、パイプラインを停止させたりすることができます。 このアプローチは、リアルタイムの品質管理が必要なストリーミング テーブルや具体化されたビューに適しています。

テーブル制約は 、Delta Lake テーブルに直接ルールを適用します。 制約により、書き込み時に無効なデータが拒否され、無効なレコードがテーブルに入力されなくなります。 このアプローチは、バッチ処理と、厳密なデータ整合性の保証を必要とするシナリオに適しています。

これらのアプローチを使用すると、データ品質の 3 つの重要なディメンションを検証できます。null 許容 (必須値が存在することを確認する)、データカーディナリティ (一意性が予想される場合の検証)、範囲チェック (値の確認は許容できる範囲内に収まります)。

null チェックを実装する

Null許容性の検証により、必要な列に値が含まれていることが確認されます。 emailとcustomer_idが常に値を持つ必要がある顧客テーブルについて考えてみましょう。 これらのチェックは、パイプラインで期待値を使用して実装できます。

Python を使用して、ストリーミング テーブル定義に期待デコレーターを追加します。

from pyspark import pipelines as dp

@dp.table
@dp.expect_or_drop("valid_email", "email IS NOT NULL")
@dp.expect_or_drop("valid_customer_id", "customer_id IS NOT NULL")
def customers():
    return spark.readStream.table("raw.customers")

SQL の同じ検証では、 CONSTRAINT 句が使用されます。

CREATE OR REFRESH STREAMING TABLE customers(
    CONSTRAINT valid_email EXPECT (email IS NOT NULL) ON VIOLATION DROP ROW,
    CONSTRAINT valid_customer_id EXPECT (customer_id IS NOT NULL) ON VIOLATION DROP ROW
) AS SELECT * FROM STREAM(raw.customers);

パイプラインの外部にある Delta Lake テーブルの場合は、テーブル定義で NOT NULL 制約を使用します。

CREATE TABLE customers (
    customer_id INT NOT NULL,
    email STRING NOT NULL,
    first_name STRING,
    last_name STRING
);

既存のテーブルに null 許容制約を追加することもできます。

ALTER TABLE customers ALTER COLUMN email SET NOT NULL;

データカーディナリティの検証

カーディナリティの検証により、列に一意の値が実際に含まれていることが保証されます。 このチェックは、主キー、トランザクション識別子、および重複がデータ品質の問題を示すその他のフィールドに不可欠です。

パイプラインの期待値は、一意性の問題を示す条件をチェックすることでカーディナリティを検証できます。 たとえば、社会保障番号が 1 人につき 1 回だけ表示されることを確認できます。

from pyspark.sql.window import Window
from pyspark.sql.functions import count

@dp.table
@dp.expect("unique_ssn_per_person", "ssn_count = 1")
def employees():
    df = spark.table("raw.employees")
    w = Window.partitionBy("ssn")
    return df.withColumn("ssn_count", count("*").over(w))

より包括的なカーディナリティ チェックを行う場合は、変換で期待値と集計ロジックを組み合わせます。

CREATE OR REFRESH MATERIALIZED VIEW order_validation AS
SELECT 
    order_id,
    COUNT(*) as occurrence_count
FROM orders
GROUP BY order_id
HAVING COUNT(*) > 1;

この具体化されたビューは、重複する注文 ID を識別し、カーディナリティの問題を調査して解決できるようにします。

Delta Lake では、予想される一意性を文書化する主キー制約がサポートされていますが、これらの制約は情報的であり、適用されません。

CREATE TABLE orders (
    order_id INT NOT NULL,
    customer_id INT,
    order_date DATE,
    CONSTRAINT orders_pk PRIMARY KEY (order_id)
);

範囲チェックを適用する

範囲の検証では、数値、日付、およびその他の値が許容範囲内にあることを確認します。 このチェックでは、データ入力エラー、システム障害、および範囲外の値を生成する統合の問題をキャッチします。

比較演算子または BETWEEN 句を使用して、範囲の期待値を定義します。

@dp.table
@dp.expect_or_fail("valid_age", "age BETWEEN 0 AND 150")
@dp.expect_or_fail("valid_salary", "salary >= 0")
@dp.expect_or_fail("valid_hire_date", "hire_date <= current_date()")
def employees():
    return spark.readStream.table("raw.employees")

SQL では、同じ範囲チェックを適用します。

CREATE OR REFRESH STREAMING TABLE transactions(
    CONSTRAINT valid_amount EXPECT (amount > 0) ON VIOLATION DROP ROW,
    CONSTRAINT valid_date EXPECT (transaction_date <= current_date()) ON VIOLATION DROP ROW,
    CONSTRAINT valid_quantity EXPECT (quantity BETWEEN 1 AND 10000) ON VIOLATION DROP ROW
) AS SELECT * FROM STREAM(raw.transactions);

Delta Lake テーブルの場合は、 CHECK 制約を使用して、書き込み時に範囲を適用します。

ALTER TABLE employees ADD CONSTRAINT valid_age CHECK (age >= 18 AND age <= 120);
ALTER TABLE transactions ADD CONSTRAINT positive_amount CHECK (amount > 0);

範囲チェックでは、複数の条件を組み合わせたビジネス ルールを検証することもできます。

@dp.expect("valid_discount", """
    discount_percent >= 0 
    AND discount_percent <= 100
    AND (discount_percent <= 50 OR customer_tier = 'PREMIUM')
""")

この期待は、割引が範囲内に留まり、プレミアム顧客のみが50%を超える割引を受けられるように強制します。

検証エラーのアクションを選択する

検証が失敗した場合は、パイプラインの応答方法を選択します。 各アクションは、さまざまなビジネス要件とデータの重要度レベルに適しています。

データ フローをブロックせずにデータ品質の問題を可視化する必要がある場合は、 警告 アクションを使用します。

@dp.expect("has_phone", "phone_number IS NOT NULL")

無効なレコードをサイレントでフィルター処理する必要がある場合はdropを使用します。

@dp.expect_or_drop("complete_address", "street IS NOT NULL AND city IS NOT NULL")

データの整合性が重要であり、無効なレコードが許容できない場合は 、失敗を使用します 。

@dp.expect_or_fail("valid_account_balance", "balance >= 0")

パイプライン UI で期待値メトリックを表示するには、期待値を含むデータセットを選択し、[ データ品質 ] タブを開きます。これらのメトリックは、検証の合格率を監視し、体系的なデータ品質の問題を特定するのに役立ちます。

検証チェックを実装する方法を理解したら、データ パイプラインで、null 値の許容、カーディナリティ、値の範囲に関する品質基準を確実に維持できます。