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

# 이 asset에 대해 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/