データ品質ツール比較：Great Expectations vs Soda vs dbt Tests でデータを検証する
オープンソースラボ編集部 ・ 2026年6月14日
データ品質ツール比較：Great Expectations vs Soda vs dbt Tests でデータを検証する
データエンジニアリングパイプラインにおいて、「データが壊れていてダッシュボードの数値が間違っていた」という事故は企業の信頼を大きく損ないます。Great Expectations（Python・柔軟なスイート定義）・Soda（YAML宣言型・Soda Cloud連携）・dbt Tests（dbt統合・SQL変換と同時にテスト）の3つがOSSデータ品質ツールのデファクトスタンダードです。
データ品質ツールが必要な理由
- サイレント障害の防止: ETLが成功しても「データの値が異常」という問題を検知するために検証が必要
- スキーマドリフト対応: ソースシステムの仕様変更（カラム型変更・NULL許容変更）をパイプラインで即座に検知
- SLAの担保: 「毎朝8時までにレポートデータを正確に届ける」というSLAをデータ品質テストで保証
- コンプライアンス: GDPR・個人情報保護のために機密データのマスキング漏れ・異常な件数変化を検知
主要ツールの概要
Great Expectations (GX)
2018年公開、Python製のOSSです。GitHubスター10k+。期待値（Expectations）を定義してデータセットを検証するツールです。300+の組み込みExpectation（expect_column_values_to_not_be_null・expect_column_values_to_be_between等）とカスタムExpectation定義・データドキュメント自動生成（HTMLレポート）が特徴です。
# Great Expectations v1.x セットアップ
pip install great_expectations
# プロジェクト初期化
gx init
# Great Expectations: PostgreSQLデータのバリデーション
import great_expectations as gx
import pandas as pd
# コンテキストを初期化
context = gx.get_context()
# データソースを追加（PostgreSQL）
datasource = context.data_sources.add_postgres(
name="production_db",
connection_string="postgresql://user:password@localhost:5432/analytics",
)
# データアセットを追加
asset = datasource.add_table_asset(
name="orders",
table_name="orders",
schema_name="public",
)
# バッチリクエストを作成（全データを対象）
batch_request = asset.build_batch_request()
# Expectation Suiteを定義
suite = context.suites.add(
gx.ExpectationSuite(name="orders_quality_suite")
)
# バリデーターを取得してExpectationを定義
validator = context.get_validator(
batch_request=batch_request,
expectation_suite=suite,
)
# NULL値チェック
validator.expect_column_values_to_not_be_null(column="order_id")
validator.expect_column_values_to_not_be_null(column="customer_id")
validator.expect_column_values_to_not_be_null(column="total_amount")
# 値の範囲チェック
validator.expect_column_values_to_be_between(
column="total_amount",
min_value=0,
max_value=10000000, # 1000万円以下
)
# 一意性チェック
validator.expect_column_values_to_be_unique(column="order_id")
# ステータス値の列挙チェック
validator.expect_column_values_to_be_in_set(
column="status",
value_set=["pending", "processing", "shipped", "delivered", "cancelled"],
)
# 件数の変化チェック（前日比で急減していないか）
validator.expect_table_row_count_to_be_between(
min_value=1000, # 最低1000件
max_value=1000000, # 最大100万件
)
# 日付カラムの形式チェック
validator.expect_column_values_to_match_strftime_format(
column="created_at",
strftime_format="%Y-%m-%d %H:%M:%S",
)
# Expectationを保存
validator.save_expectation_suite()
# バリデーション実行
results = context.run_validation_definition(
validation_definition=context.validation_definitions.add(
gx.ValidationDefinition(
name="orders_daily_check",
data=asset.build_batch_request(),
suite=suite,
)
)
)
print(f"バリデーション成功: {results.success}")
print(f"テスト数: {results.statistics['evaluated_expectations']}")
print(f"失敗数: {results.statistics['unsuccessful_expectations']}")
# 失敗したテストの詳細を表示
for result in results.results:
if not result.success:
print(f" ❌ {result.expectation_config.expectation_type}")
print(f" 対象: {result.expectation_config.kwargs.get('column')}")
print(f" 詳細: {result.result}")
# Great Expectations: Pandasデータフレームの検証（CI/CDで軽量テスト）
import great_expectations as gx
import pandas as pd
df = pd.read_csv("data/daily_sales.csv")
gx_df = gx.from_pandas(df)
result = gx_df.expect_column_values_to_not_be_null("sale_amount")
assert result.success, f"NULLチェック失敗: {result}"
result = gx_df.expect_column_values_to_be_between(
"sale_amount", min_value=0, max_value=1_000_000
)
assert result.success, f"金額範囲チェック失敗: {result}"
print("✅ データ品質チェック通過")
Soda
2020年公開、Python製のOSSです。GitHubスター1.5k+。YAML形式でデータ品質チェックを宣言的に記述するツールです。Soda Cloud（有料）と連携してチームでデータ品質を可視化できますが、soda-core（OSSコア）はCLIとPythonで完全無料で使えます。
# Soda Core のインストール（PostgreSQL向け）
pip install soda-core-postgres
# チェックをスキャン実行
soda scan -d production_db -c ./soda/configuration.yml ./soda/checks/orders.yml
# soda/configuration.yml - データソース設定
data_sources:
production_db:
type: postgres
host: localhost
port: 5432
username: ${DB_USER}
password: ${DB_PASSWORD}
database: analytics
schema: public
# soda/checks/orders.yml - SodaCL（Soda Check Language）でチェックを定義
checks for orders:
# テーブル存在確認
- schema:
name: ordersテーブルスキーマチェック
warn:
when required column missing: [order_id, customer_id, total_amount, created_at]
fail:
when wrong column type:
order_id: bigint
total_amount: numeric
# NULL値チェック
- missing_count(order_id) = 0:
name: order_idにNULLなし
- missing_count(customer_id) = 0:
name: customer_idにNULLなし
# 件数チェック（前日比アノマリー検出）
- anomaly score for row_count < default:
name: 件数のアノマリー検出
# 重複チェック
- duplicate_count(order_id) = 0:
name: order_id重複なし
# 値の範囲チェック
- invalid_count(total_amount):
valid min: 0
valid max: 10000000
name: 合計金額が0〜1000万円の範囲内
# ステータス値チェック
- invalid_count(status) = 0:
valid values: [pending, processing, shipped, delivered, cancelled]
name: ステータス値が有効な選択肢内
# 鮮度チェック（最新データが12時間以内）
- freshness(created_at) < 12h:
name: データ鮮度チェック（12時間以内）
checks for customers:
- missing_count(email) = 0
- missing_percent(phone) < 10%:
name: 電話番号の欠損率が10%未満
dbt Tests
2016年公開（dbt）、Python製のOSSです。GitHubスター10k+。dbtのデータ変換（SQL モデル）と同じYAML・SQLファイルでテストを定義できます。dbt transformsとテストが一体化しているため、データウェアハウス（BigQuery・Snowflake・Redshift・DuckDB）で使うデータエンジニアリングチームに最適です。
# models/schema.yml - dbt YAMLテスト定義
version: 2
models:
- name: stg_orders
description: "注文テーブルのステージングモデル"
columns:
- name: order_id
description: "注文ID（主キー）"
tests:
- unique # 一意性テスト
- not_null # NOT NULLテスト
- name: customer_id
tests:
- not_null
- relationships: # 参照整合性テスト
to: ref('stg_customers')
field: customer_id
- name: status
tests:
- accepted_values: # 許可された値のみ
values: ['pending', 'processing', 'shipped', 'delivered', 'cancelled']
- name: total_amount
tests:
- not_null
- dbt_utils.expression_is_true: # dbt-utilsパッケージの式テスト
expression: ">= 0"
- name: fct_daily_sales
description: "日次売上ファクトテーブル"
tests:
- dbt_utils.recency: # データ鮮度テスト
datepart: hour
field: created_at
interval: 24
columns:
- name: sale_date
tests:
- unique
- not_null
- name: revenue
tests:
- not_null
- dbt_utils.expression_is_true:
expression: "> 0"
-- tests/assert_orders_not_negative_amount.sql
-- カスタムテスト: 金額がマイナスの注文が存在しないことを確認
-- dbt testが実行してレコードが返ればテスト失敗
SELECT
order_id,
total_amount
FROM {{ ref('stg_orders') }}
WHERE total_amount < 0
# Airflow DAG: Great Expectations + dbt を組み合わせたパイプライン
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta
def run_gx_validation(**kwargs):
import great_expectations as gx
context = gx.get_context()
# バリデーションを実行してAlertを送信
results = context.run_checkpoint(checkpoint_name="daily_ingestion_check")
if not results.success:
raise Exception(f"データ品質チェック失敗: {results.statistics}")
with DAG(
'data_quality_pipeline',
schedule_interval='0 6 * * *', # 毎朝6時
start_date=datetime(2024, 1, 1),
default_args={'retries': 2, 'retry_delay': timedelta(minutes=5)},
) as dag:
# 1. データ取り込み後にGXでソースデータを検証
gx_source_check = PythonOperator(
task_id='gx_source_validation',
python_callable=run_gx_validation,
)
# 2. dbt transformsを実行
dbt_run = BashOperator(
task_id='dbt_run',
bash_command='dbt run --project-dir /opt/dbt/analytics --profiles-dir /opt/dbt',
)
# 3. dbt testsを実行（変換後のデータを検証）
dbt_test = BashOperator(
task_id='dbt_test',
bash_command='dbt test --project-dir /opt/dbt/analytics --profiles-dir /opt/dbt',
)
# 4. Sodaで鮮度・アノマリーを最終確認
soda_scan = BashOperator(
task_id='soda_final_scan',
bash_command='soda scan -d prod_warehouse -c /opt/soda/config.yml /opt/soda/checks/marts.yml',
)
gx_source_check >> dbt_run >> dbt_test >> soda_scan
機能比較表
| 比較項目 | Great Expectations | Soda | dbt Tests |
|---|---|---|---|
| 設定形式 | Python | YAML | YAML/SQL |
| dbt統合 | △ | ✅ | ✅（同一ツール） |
| アノマリー検出 | △ | ✅ | ❌ |
| HTMLレポート | ✅ | ✅（Soda Cloud） | ✅（dbt docs） |
| 学習コスト | 中 | 低 | 低（dbt既存） |
データ品質ツールはDevOpsカテゴリ/categories/devopsのApache Airflow・Prefect・dbtパイプラインに組み込んでETL完了後の自動品質ゲートを設置します。LLM Toolsカテゴリ/categories/llm-toolsのベクターデータベースに入力するRAGの知識データの品質検証にも活用されています。
FAQ
Q. Great ExpectationsとSodaのどちらを選べばいいですか？
A. Pythonコードでの高度なカスタムバリデーション・HTMLレポートが必要ならGreat Expectations、YAMLで素早く宣言的に定義・dbtと組み合わせるならSodaが向いています。Great Expectationsの強み: ①300+の組み込みExpectationで複雑な条件も定義可能②DataDocsでHTMLレポートを自動生成して非エンジニアにも共有③カスタムExpectation（Pythonクラス）で独自ロジックを追加可能。Sodaの強み: ①SodaCL（YAML）で設定が直感的・可読性が高い②Soda Cloudでチームダッシュボード・アラート管理③anomaly scoreチェックで機械学習ベースのアノマリー検出（Soda Cloudが必要）。
Q. dbtを使っていない場合でもdbt Testsは使えますか？
A. dbt Testsはdbt projectの一部なのでdbt単独では使えませんが、dbtを導入することで同時にデータ変換とデータ品質の両方を管理できます。dbt Core（OSS）は無料で、PostgreSQL・BigQuery・Snowflake・Redshift・DuckDB等のデータウェアハウスに対してSQLモデルを実行できます。「まずdbt Testsだけ使いたい」場合は、dbt modelを1つも書かずにsource（ソーステーブルのメタデータ）に対してのみtestsを定義することも可能です（dbt test --select source:*）。
Q. データパイプラインのどの時点でデータ品質チェックを実施すべきですか？
A. 3つのタイミングが推奨されます。①ソースデータの取り込み直後: GXやSodaでソーステーブルのスキーマ・NULL・重複・件数を検証（データソースの品質問題を早期に検知）②dbt変換後（Staging→Mart）: dbt Testsで参照整合性・accepted_values・uniquenessを検証（変換ロジックのバグを検知）③下流への配信前（BI/API接続前）: Sodaの鮮度チェック・アノマリー検出でファイナルゲートを設置（エンドユーザーに不正確なデータを届けない）。失敗時の通知: Slack Webhookへのアラート送信（GX AlertDestination・Soda notification）でデータエンジニアに即座に通知します。
Q. カラムの分布変化（ドリフト）を自動検知するにはどうしますか？
A. ①Great Expectations: expect_column_mean_to_be_between・expect_column_stdev_to_be_betweenで統計的な閾値チェック。またはProfilerRuleでデータプロファイリングを実行してベースライン統計を自動生成②Soda: anomaly score for column_metric < defaultで機械学習ベースの自動アノマリー検出（要Soda Cloud）。OSSのみで使う場合はmin・max・avgチェックを組み合わせ③Monte Carlo Data / Bigeye: 本格的なデータ可観測性プラットフォーム（商用）でMLベースの自動ドリフト検出④Evidently AI: MLモデルへの入力データのドリフト検出に特化したOSS（pip install evidently）。
まとめ
| ユースケース | 推奨ツール |
|---|---|
| Python・高度なカスタム検証・HTMLレポート | Great Expectations |
| YAML宣言型・dbt連携・素早い導入 | Soda |
| dbt既存ユーザー・SQL変換と一体管理 | dbt Tests |