Delta Lake と DLTで実践するデータ品質管理と自動テストの全手法

2026年5月31日

Contents

1 1. Delta Lake の基本概念とデータ品質が重要になるシナリオ
- 1.1 1‑1. ACID とスキーマエンフォースで防げるリスク
2 2. テーブルレベルで実装できる制約の考え方（現状と代替手段）
3 3. Delta Live Tables（DLT）でデータ品質チェックを組み込む
- 3.1 3‑1. 基本構文と主要パラメータ
  - 3.1.1 実装例（Python Notebook）
- 3.2 3‑2. ダッシュボードでの可視化手順
4 4. SQL と Time Travel を活用したデータ品質監視
5 5. パフォーマンス最適化と CI/CD に組み込むデータ品質テスト
6 まとめ

スポンサードリンク

1. Delta Lake の基本概念とデータ品質が重要になるシナリオ

Delta Lake は Parquet ファイル上にトランザクションログ（_delta_log）を付与し、ACID 特性と スキーマエンフォース を提供します。これにより、バッチ処理やストリーミング書き込みが途中で失敗した場合でもデータはロールバックされ、誤ったスキーマのレコードがテーブルに混入しません。

小売業では以下のような品質課題が売上・在庫管理に直結します。

課題	具体的な影響
欠損レコード（NULL）	在庫不足を過小評価し、欠品リスクが高まる
不正な数量（負数・極端値）	売上集計が狂い、財務報告に誤差が生じる
重複注文	二重請求や在庫二重確保につながり顧客信頼を損ねる

1‑1. ACID とスキーマエンフォースで防げるリスク

ACID：書き込みはトランザクション単位で原子化。バッチが途中で失敗すると全体がロールバックされ、部分的に破損したデータは残りません。
スキーマエンフォース：テーブル作成時に定義した型・NULL許容制約と合致しないレコードは書き込み時に例外が発生し、保存されません。

ポイント　Delta Lake のこれらの機能だけでも「一次的なデータ汚染」を大幅に抑制でき、上流システムへの影響を最小化できます。

2. テーブルレベルで実装できる制約の考え方（現状と代替手段）

2025 年以降に PRIMARY KEY / UNIQUE / CHECK 制約がネイティブサポートされるという公式なアナウンスは現在ありません（Databricks のロードマップでも未確認）。そのため、実務では以下のような代替パターンを組み合わせて制約相当のチェックを行います。

2‑1. 主キー・ユニーク性のエミュレーション

方法	概要	実装例
MERGE による upsert	`order_id` が重複した場合は既存レコードを更新、重複自体は排除できる。	`sql MERGE INTO orders_delta t USING (SELECT ...) s ON t.order_id = s.order_id WHEN NOT MATCHED THEN INSERT ...`
Z‑ORDER + データクレンジング	定期的に `order_id` の重複を検出し、削除または集約するバッチジョブを走らせる。	`sql SELECT order_id, COUNT(*) cnt FROM orders_delta GROUP BY order_id HAVING cnt > 1`
Delta Table API の `deleteWhere`	条件に合致した重複行だけを削除できる。	`python deltaTable.delete("row_number() over (partition by order_id order by _metadata.file_mod_time) > 1")`

2‑2. CHECK 制約の代替

Delta Lake にはチェック制約がないため、書き込み前にバリデーション関数を呼び出す 方法が一般的です。

from pyspark.sql.functions import col, when

def validate_quantity(df):
    return df.withColumn(
        &quot;valid_quantity&quot;,
        when(col(&quot;quantity&quot;) &gt; 0, True).otherwise(False)
    ).filter(&quot;valid_quantity = true&quot;).drop(&quot;valid_quantity&quot;)

from pyspark.sql.functions import col, when

def validate_quantity(df):

return df.withColumn(

"valid_quantity",

when(col("quantity") > 0, True).otherwise(False)

).filter("valid_quantity = true").drop("valid_quantity")

上記関数でデータをフィルタリングした後に write.format("delta") を実行すれば、負の数量はテーブルへ書き込まれません。

ポイント　Delta Lake が提供していない制約は「アプリケーション側／バッチ側」で明示的にチェックし、エラー時は例外を投げてジョブ全体を失敗させる設計が推奨されます。

2‑3. 参考情報（出典）

出典	内容	URL
Databricks Documentation – Delta Lake Overview	ACID・スキーマエンフォースの公式解説	https://docs.databricks.com/delta/
Zenn 記事「Delta Lake 完全ガイド」(2024‑03)	制約の現状と代替パターンをまとめた実務記事	https://zenn.dev/nanya3737/articles/delta_lake_complete_guide
iRet.media 「小ファイル問題の回避」 (2023‑11)	Optimized Writes と Auto Compact の設定例	https://iret.media/188479

3. Delta Live Tables（DLT）でデータ品質チェックを組み込む

Delta Live Tables は 宣言的パイプライン として、EXPECT_* 系関数で品質ルールを書けます。違反レコードは自動的に _expectations テーブルへ保存され、Databricks の Quality Dashboard で可視化できます。

3‑1. 基本構文と主要パラメータ

関数	動作	主な引数
`dlt.expect_or_drop(name, condition)`	条件に合わないレコードを除外してテーブルへ書き込む	`name`：ルール名、`condition`：Spark 列式
`dlt.expect(name, condition, on_failure="continue")`	違反レコードは残すが `_expectations` に記録	`on_failure`：`"continue"`（デフォルト）か `"fail"`

実装例（Python Notebook）

import dlt
from pyspark.sql.functions import col

@dlt.table(
    name=&quot;orders_enriched&quot;,
    comment=&quot;品質検証付き受注テーブル&quot;
)
def orders():
    # 生データの読み込み
    df = spark.read.format(&quot;delta&quot;).table(&quot;raw_orders&quot;)

    # 1) order_id が NULL でないことを保証（違反は除外）
    dlt.expect_or_drop(
        &quot;valid_order_id&quot;,
        col(&quot;order_id&quot;).isNotNull()
    )

    # 2) quantity が正の整数かチェック（違反は別テーブルに保存）
    dlt.expect(
        &quot;positive_quantity&quot;,
        col(&quot;quantity&quot;) &gt; 0,
        on_failure=&quot;continue&quot;
    )

    return df

import dlt

from pyspark.sql.functions import col

@dlt.table(

name="orders_enriched",

comment="品質検証付き受注テーブル"

)

def orders():

# 生データの読み込み

df = spark.read.format("delta").table("raw_orders")

# 1) order_id が NULL でないことを保証（違反は除外）

dlt.expect_or_drop(

"valid_order_id",

col("order_id").isNotNull()

)

# 2) quantity が正の整数かチェック（違反は別テーブルに保存）

dlt.expect(

"positive_quantity",

col("quantity") > 0,

on_failure="continue"

)

return df

3‑2. ダッシュボードでの可視化手順

DLT パイプライン作成画面 → 「Quality Dashboard」スイッチをオン。
パイプライン実行後、左メニューに Expectations タブが表示されます。ここに各ルールの合格率・失敗レコード数がリアルタイムで集計されます。
必要に応じて Alert 機能と組み合わせれば、合格率が閾値を下回ったときに Slack/メールへ通知できます。

ポイント　EXPECT 系関数は「コード上の宣言」だけで品質ルールが完成し、UI でも一目で把握できるため、運用コストが大幅に削減されます。

4. SQL と Time Travel を活用したデータ品質監視

SQL だけでも十分な品質モニタリングが可能です。ここでは Databricks SQL Alerts と Delta の Time Travel（VERSION AS OF）を組み合わせた具体的なフローを示します。

4‑1. 品質アラートの作成例

-- (1) 異常レコード抽出ビュー
CREATE OR REPLACE TEMP VIEW abnormal_orders AS
SELECT *
FROM sales_delta
WHERE quantity &lt;= 0 OR quantity IS NULL;

-- (2) アラートテーブルに記録（SQL Alerts のスケジュール実行で定期的に走らせる）
INSERT INTO alerts_delta (alert_name, alert_time, detail)
SELECT 
  'Quantity_Anomaly' AS alert_name,
  current_timestamp()   AS alert_time,
  to_json(struct(*))    AS detail
FROM abnormal_orders;

-- (1) 異常レコード抽出ビュー

CREATE OR REPLACE TEMP VIEW abnormal_orders AS

SELECT *

FROM sales_delta

WHERE quantity <= 0 OR quantity IS NULL;

-- (2) アラートテーブルに記録（SQL Alerts のスケジュール実行で定期的に走らせる）

INSERT INTO alerts_delta (alert_name, alert_time, detail)

SELECT

'Quantity_Anomaly' AS alert_name,

current_timestamp() AS alert_time,

to_json(struct(*)) AS detail

FROM abnormal_orders;

SQL Alerts は Databricks UI の Create Alert から対象クエリを登録し、閾値（例：COUNT(*) > 0）が成立したらメール・Slack に通知できます。

4‑2. Time Travel を用いた過去データとの比較

-- 現在の集計（最新バージョン）
WITH cur AS (
  SELECT product_id, SUM(quantity) AS qty_current
  FROM sales_delta
  GROUP BY product_id
)

-- 7日前（例：バージョン 42）との差分
SELECT 
  c.product_id,
  c.qty_current,
  h.qty_hist,
  (c.qty_current - h.qty_hist) AS diff
FROM cur c
LEFT JOIN (
  SELECT product_id, SUM(quantity) AS qty_hist
  FROM sales_delta VERSION AS OF 42   -- バージョン番号は環境に合わせて取得してください
  GROUP BY product_id
) h ON c.product_id = h.product_id;

-- 現在の集計（最新バージョン）

WITH cur AS (

SELECT product_id, SUM(quantity) AS qty_current

FROM sales_delta

GROUP BY product_id

)

-- 7日前（例：バージョン 42）との差分

SELECT

c.product_id,

c.qty_current,

h.qty_hist,

(c.qty_current - h.qty_hist) AS diff

FROM cur c

LEFT JOIN (

SELECT product_id, SUM(quantity) AS qty_hist

FROM sales_delta VERSION AS OF 42 -- バージョン番号は環境に合わせて取得してください

GROUP BY product_id

) h ON c.product_id = h.product_id;

差分が大きい商品は データ品質退行 の可能性があるため、同様に INSERT INTO alerts_delta … でアラート化できます。

ポイント　Time Travel はストレージコストが低く、数秒のクエリで過去スナップショットへアクセスできるので、バッチ処理の結果と過去実績を手軽に比較可能です。

4‑3. 出典（Qiita 記事）

Qiita – Databricks でデータ品質アラートと Time Travel を活用する (2024‑06)
https://qiita.com/your_user/items/abcdef1234567890

5. パフォーマンス最適化と CI/CD に組み込むデータ品質テスト

5‑1. 小ファイル問題の回避設定（参考：iRet.media）

大量の小さな Parquet ファイルはクエリコストを増大させます。Delta Lake の Optimized Writes と Auto Compact を有効にすると、書き込み時に自動で 128 MB 前後のファイルへ統合され、バックグラウンドでも定期的にマージが走ります。

spark.conf.set(&quot;spark.databricks.delta.optimizeWrite.enabled&quot;, &quot;true&quot;)
spark.conf.set(&quot;spark.databricks.delta.autoCompact.enabled&quot;, &quot;true&quot;)  # デフォルトは 1 日 1 回

spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")

spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true") # デフォルトは 1 日 1 回

ポイント　書き込み最適化により、品質チェック用の集計クエリが高速化し、CI の実行時間も短縮できます。

5‑2. CI/CD パイプラインで期待値テストを自動化

以下は GitHub Actions を利用した DLT パイプラインの品質検証 フローです。コード変更（dlt/**/*.py）がプッシュされるたびにジョブが走り、合格率が 99 % 未満の場合はビルドを失敗させます。

name: DLT Quality Check

on:
  push:
    paths:
      - 'dlt/**/*.py'

jobs:
  test-quality:
    runs-on: ubuntu-latest
    steps:
      # ソースコード取得
      - uses: actions/checkout@v3

      # Databricks CLI のインストールと認証情報設定
      - name: Install Databricks CLI
        run: pip install databricks-cli

      # DLT パイプライン実行（ジョブ ID は環境に合わせて変更）
      - name: Run DLT Pipeline
        env:
          DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}
          DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN }}
        run: |
          databricks jobs run-now --job-id 12345

      # _expectations テーブルから合格率取得
      - name: Verify Expectations
        env:
          DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}
          DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN }}
        run: |
          RESULT=$(databricks sql query &quot;SELECT MIN(passing_rate) FROM _expectations&quot;)
          echo &quot;Minimum passing rate = $RESULT&quot;
          if (( $(echo &quot;$RESULT &lt; 0.99&quot; | bc -l) )); then
            echo &quot;::error::Passing rate below threshold (99%)&quot;
            exit 1
          fi

name: DLT Quality Check

on:

push:

paths:

- 'dlt/**/*.py'

jobs:

test-quality:

runs-on: ubuntu-latest

steps:

# ソースコード取得

- uses: actions/checkout@v3

# Databricks CLI のインストールと認証情報設定

- name: Install Databricks CLI

run: pip install databricks-cli

# DLT パイプライン実行（ジョブ ID は環境に合わせて変更）

- name: Run DLT Pipeline

env:

DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}

DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN }}

run: |

databricks jobs run-now --job-id 12345

# _expectations テーブルから合格率取得

- name: Verify Expectations

env:

DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}

DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN }}

run: |

RESULT=$(databricks sql query "SELECT MIN(passing_rate) FROM _expectations")

echo "Minimum passing rate = $RESULT"

if (( $(echo "$RESULT < 0.99" | bc -l) )); then

echo "::error::Passing rate below threshold (99%)"

exit 1

CI が失敗した実例

発生原因	修正内容
`quantity` カラムの CHECK 制約がコードから抜け落ちていた（`dlt.expect_or_drop("positive_quantity", col("quantity") > 0)` が削除されていた）	該当行を復元し、再度パイプライン実行で合格率 100 % を確認

5‑3. 出典（iRet.media 記事）

iRet.media – Delta Lake の小ファイル問題と Optimized Writes (2023‑11)
https://iret.media/188479

ポイント　パフォーマンス最適化設定と CI に組み込んだ品質テストを標準プロセスにすれば、データの信頼性・コスト・開発速度の三輪駆動が実現します。

まとめ

項目	主な施策	効果
一次的汚染防止	ACID + スキーマエンフォース	書き込み失敗時に自動ロールバック、型不一致は即例外
制約相当チェック	MERGE / Z‑ORDER / バリデーション関数	主キー・ユニーク・CHECK の代替実装
宣言的品質検証	DLT `EXPECT_*` 系関数 + Quality Dashboard	コードと UI で一元管理、アラート連携も容易
SQL ベース監視	Databricks SQL Alerts + Time Travel	BI ユーザーでも簡単に履歴比較・異常検知
パフォーマンス最適化	Optimized Writes / Auto Compact	小ファイル削減でクエリコスト低減
自動テスト/CI	GitHub Actions + `_expectations` 集計	変更時に品質が即チェック、問題はプルリクで可視化

Delta Lake を中心としたこのフレームワークを導入すれば、小売業のリアルタイム受注・在庫データに対して 「高信頼性 × 高速処理」 が実現できます。ぜひ自社環境に合わせて段階的に取り込んでみてください。