Great Expectations 2026年版：データ品質検証とインタビュー対策

# gx_setup.py
import great_expectations as gx

# 既存のData Contextを初期化またはロード
context = gx.get_context()

# ローカルファイル用のPandas Data Sourceを追加
data_source = context.data_sources.add_pandas("local_files")

# 特定のCSVパターンを指すData Assetを定義
data_asset = data_source.add_csv_asset(
    name="user_events",
    filepath_or_buffer="data/user_events_*.csv"  # Globパターン
)

# build_suite.py
import great_expectations as gx

context = gx.get_context()

# Expectation Suiteを作成または取得
suite = context.suites.add(
    gx.ExpectationSuite(name="user_events_suite")
)

# Suiteにexpectationsを追加
suite.add_expectation(
    gx.expectations.ExpectColumnToExist(column="user_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="user_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="age",
        min_value=0,
        max_value=120
    )
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToMatchRegex(
        column="email",
        regex=r"^[\w.-]+@[\w.-]+\.\w+$"
    )
)

# SuiteをData Contextに保存
context.suites.save(suite)

# run_checkpoint.py
import great_expectations as gx

context = gx.get_context()

# Checkpointを作成
checkpoint = context.checkpoints.add(
    gx.Checkpoint(
        name="user_events_checkpoint",
        validation_definitions=[
            gx.ValidationDefinition(
                name="validate_user_events",
                data=context.data_sources.get("local_files")
                    .get_asset("user_events")
                    .build_batch_request(),
                suite=context.suites.get("user_events_suite")
            )
        ],
        actions=[
            gx.checkpoint.UpdateDataDocsAction(name="update_docs"),
        ]
    )
)

# Checkpointを実行
result = checkpoint.run()

# 全体の成功を確認
if result.success:
    print("すべての検証に成功しました")
else:
    print("検証の失敗が検出されました")
    for validation_result in result.run_results.values():
        for expectation_result in validation_result.results:
            if not expectation_result.success:
                print(f"  失敗: {expectation_result.expectation_config}")

# dags/user_events_pipeline.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import great_expectations as gx

def validate_user_events():
    """ユーザーイベントデータに対してGX Checkpointを実行します。"""
    context = gx.get_context(context_root_dir="/opt/airflow/gx")
    checkpoint = context.checkpoints.get("user_events_checkpoint")
    result = checkpoint.run()
    
    if not result.success:
        # Airflowタスクを失敗させるために例外を発生させる
        raise ValueError("データ検証に失敗しました。詳細はData Docsを確認してください。")

with DAG(
    dag_id="user_events_pipeline",
    start_date=datetime(2026, 1, 1),
    schedule_interval="@daily",
    catchup=False
) as dag:
    
    validate_task = PythonOperator(
        task_id="validate_user_events",
        python_callable=validate_user_events
    )
    
    # 下流のタスクは検証の成功に依存する
    # transform_task >> load_task

# custom_expectations/expect_valid_iso_country_code.py
from great_expectations.expectations import Expectation
from great_expectations.core import ExpectationConfiguration
import pycountry

class ExpectValidISOCountryCode(Expectation):
    """有効なISO 3166-1 alpha-2国コードを検証します。"""
    
    column: str
    
    def _validate(
        self,
        metrics: dict,
        runtime_configuration: dict = None,
        execution_engine = None
    ):
        column_values = metrics.get("column_values.value_set")
        
        invalid_codes = []
        for code in column_values:
            if code is not None:
                try:
                    pycountry.countries.get(alpha_2=code.upper())
                except KeyError:
                    invalid_codes.append(code)
        
        success = len(invalid_codes) == 0
        return {
            "success": success,
            "result": {
                "observed_value": f"{len(invalid_codes)} invalid codes",
                "invalid_codes": invalid_codes[:10]  # 最初の10件を表示
            }
        }

# snowflake_validation.py
import great_expectations as gx

context = gx.get_context()

# Snowflake Data Sourceを追加
snowflake_source = context.data_sources.add_snowflake(
    "snowflake_warehouse",
    connection_string=(
        "snowflake://user:password@account/"
        "database/schema?warehouse=compute_wh"
    )
)

# クエリベースのData Assetを追加
orders_asset = snowflake_source.add_query_asset(
    name="recent_orders",
    query="SELECT * FROM orders WHERE order_date >= CURRENT_DATE - 7"
)

# このアセットに対してExpectationを適用
suite = context.suites.get("orders_suite")
validation_definition = gx.ValidationDefinition(
    name="validate_recent_orders",
    data=orders_asset.build_batch_request(),
    suite=suite
)

result = validation_definition.run()

# .github/workflows/data-validation.yml
name: Data Validation

on:
  push:
    branches: [main]
  schedule:
    - cron: '0 */6 * * *'  # 6時間ごと

jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      
      - name: Install dependencies
        run: |
          pip install great-expectations[snowflake]==1.22.0
      
      - name: Run GX Checkpoints
        env:
          SNOWFLAKE_PASSWORD: ${{ secrets.SNOWFLAKE_PASSWORD }}
        run: |
          python scripts/run_all_checkpoints.py
      
      - name: Upload Data Docs
        uses: actions/upload-artifact@v4
        with:
          name: data-docs
          path: gx/uncommitted/data_docs/