データ型チェックを実装する

データセット内のすべての列には予想されるデータ型があり、受信データがその期待値と一致しない場合、問題は下流に連鎖します。 整数が属する文字列値は、計算を中断したり、集計が破損したり、パイプラインエラーが発生したりする可能性があります。 データ型チェックでは、テーブルに入力する前に、各フィールドに正しい型の値が含まれていることが確認されます。

このユニットでは、Azure Databricks でスキーマの強制、明示的な型キャスト、検証の制約を使用してデータ型チェックを実装する方法について説明します。

スキーマ適用の理解

Delta Lake では、テーブルにデータを書き込むと、データ型が自動的に検証されます。 スキーマ 強制と呼ばれるこの組み込みメカニズムは、受信列のデータ型をターゲット テーブルのスキーマと比較します。 型の不一致が発生すると、Delta Lake は値を安全にキャストしようとします。 キャストが失敗すると、書き込み操作でエラーが発生します。

スキーマの適用では、挿入操作中に次の規則が適用されます。

• 挿入されたすべての列がターゲット テーブルに存在する必要があります
• すべての列データ型は、ターゲット型と一致するか、安全にキャストできる必要があります

整数列を持つテーブルについて考えてみましょう。

CREATE TABLE inventory (
    product_id INT,
    quantity INT,
    last_updated DATE
);

quantityが数値を表す文字列であるデータを挿入すると、Delta Lake によってターゲットの型にキャストされます。

-- This succeeds because '100' can be cast to INT
INSERT INTO inventory VALUES (1, '100', '2026-01-15');

ただし、整数に変換できない文字列を挿入すると、操作が失敗します。

-- This fails because 'fifty' cannot be cast to INT
INSERT INTO inventory VALUES (2, 'fifty', '2026-01-15');

スキーマの適用により、型の不一致に対する防御の第 1 行が提供されます。 不一致の処理方法をより詳細に制御するには、明示的なキャストを使用できます。

明示的な型キャストを使用する

cast関数は、あるデータ型から別のデータ型に値を変換します。 変換に失敗すると、エラーが発生します。 try_cast関数は同じように動作しますが、変換が失敗したときにエラーを発生させる代わりにNULLを返します。

無効なデータの処理を停止する厳密な検証が必要な場合は、 cast を使用します。

SELECT 
    cast(raw_amount AS DECIMAL(10,2)) AS amount,
    cast(raw_date AS DATE) AS transaction_date
FROM staging_data;

クエリに失敗せずに無効な値を識別する場合は、 try_cast を使用します。

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;

try_castでは、無効な値を持つレコードは、キャスト列のNULLを返します。 その後、データ品質要件に基づいて、これらのレコードをフィルター処理、フラグ設定、または検疫できます。

制約を使用して型検証を実装する

CHECK 制約 を使用すると、データが挿入または更新されるたびに適用されるカスタム検証規則を定義できます。 通常、値の範囲とパターンに使用されますが、型対応関数と組み合わせて高度な型チェックを作成できます。

たとえば、文字列列に数値のみが含まれていることを検証できます。

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]+)?$')

この制約により、 order_total 列に有効な数値のような値が含まれていることが保証され、テーブルに入る前に正しくないデータがキャッチされます。

日付の検証では、制約内で try_cast 関数を使用できます。

ALTER TABLE events 
ADD CONSTRAINT valid_event_date CHECK (try_cast(event_date_str AS DATE) IS NOT NULL);

この方法では、 event_date_str に有効な日付形式が含まれていないレコードが拒否されます。

パイプラインにおける型不一致の処理

外部ソースからのデータを処理する場合、型の不一致が一般的です。 有効なレコードと無効なレコードを分離するパターンを実装します。これにより、レビューのために問題のあるレコードを検疫しながら、適切なデータを処理できます。

-- 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;

このパターンにより、後で分析するために無効なレコードを保持しながら、パイプラインで有効なデータの処理が続行されます。

型チェックにパイプラインの期待値を使用する

Lakeflow パイプラインでは、パイプライン定義でデータ品質ルールを直接定義できる expectations が提供されます。 期待値を使用して、値を予期される型にキャストできることを確認できます。

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

expect_or_dropでは、型チェックに失敗したレコードは、ターゲット テーブルに到達する前に削除されます。 expectを使用して、レコードを削除せずに違反をログに記録するか、違反が発生したときにパイプラインを停止expect_or_fail。

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

パイプラインの期待は、データ型の検証を ETL ロジックに直接統合し、パイプライン UI を介してデータ品質メトリックを可視化します。

スキーマの適用、明示的なキャスト、制約、パイプラインの期待など、複数のレベルでデータ型チェックを実装すると、多層防御が作成されます。 各レイヤーは、他のレイヤーを超える可能性がある問題をキャッチし、Unity カタログ テーブルのデータ品質を向上させます。