Great Expectations 2026: การตรวจสอบคุณภาพข้อมูลและคำถามสัมภาษณ์

Great Expectations (GX) เป็น framework โอเพนซอร์สมาตรฐานสำหรับการตรวจสอบคุณภาพข้อมูลใน data pipeline ที่ใช้ Python เป็นฐาน เวอร์ชัน 1.22 ที่เปิดตัวในเดือนสิงหาคม 2026 ได้รวบรวมการเปลี่ยนแปลง API ที่เริ่มใช้ใน GX 1.0 และเพิ่มการสนับสนุน Python 3.14 แบบทดลอง ทำให้เป็นเครื่องมือสำคัญสำหรับ data engineer ที่สร้าง pipeline ระดับ production

ปัญหาที่ Great Expectations แก้ไขใน Data Pipeline

Data pipeline มักจะล้มเหลวอย่างเงียบ ๆ การเปลี่ยนแปลง schema ที่ต้นทาง ค่า null ที่ไม่ควรมี รูปแบบวันที่ที่เปลี่ยนจาก ISO เป็น Unix timestamp: ปัญหาเหล่านี้มักจะไปถึงแดชบอร์ดหรือโมเดล ML ก่อนที่ใครจะสังเกตเห็น Great Expectations ปฏิบัติต่อข้อมูลเหมือนโค้ด โดยใช้ assertion (เรียกว่า Expectation) ที่ทำงานอัตโนมัติที่ checkpoint ใน pipeline

Framework นี้ผสานกับ Apache Airflow, Databricks, Snowflake และบริการ cloud storage เช่น AWS S3 และ Azure Blob Storage การตรวจสอบแต่ละครั้งจะสร้าง Data Docs ซึ่งเป็นรายงาน HTML ที่ stakeholder ที่ไม่ใช่ฝ่ายเทคนิคสามารถอ่านได้

แนวคิดหลัก: Data Context, Data Sources และ Expectations

GX 1.22 จัดระเบียบการตรวจสอบรอบ ๆ สี่องค์ประกอบ: Data Context, Data Sources, Data Assets และ Expectation Suites

Data Context เป็นออบเจ็กต์การกำหนดค่าส่วนกลาง มันเก็บ metadata สำหรับ Data Sources, Expectation Suites, Checkpoints และ Validation Results ในอดีต ในโปรเจกต์ส่วนใหญ่ ไดเรกทอรี gx/ เดียวจะเก็บไฟล์การกำหนดค่า YAML และ Data Docs ที่สร้างขึ้น

Data Source แสดงถึงการเชื่อมต่อกับ database, data warehouse หรือ file system ส่วน Data Asset เป็นชุดของ record ที่มีตรรกะภายใน source นั้น คล้ายกับตารางหรือ result set ของ query

# gx_setup.py
import great_expectations as gx

# Initialize or load an existing Data Context
context = gx.get_context()

# Add a Pandas Data Source for local files
data_source = context.data_sources.add_pandas("local_files")

# Define a Data Asset pointing to a specific CSV pattern
data_asset = data_source.add_csv_asset(
    name="user_events",
    filepath_or_buffer="data/user_events_*.csv"  # Glob pattern
)

การตั้งค่านี้ช่วยให้ GX สามารถตรวจสอบ CSV ใด ๆ ที่ตรงกับ user_events_*.csv ในไดเรกทอรี data/

การสร้าง Expectation Suite สำหรับการตรวจสอบคอลัมน์

Expectation Suite เป็นชุดของ assertion ต่อ Data Asset แต่ละ Expectation ประกาศเงื่อนไขที่ควรเป็นจริงสำหรับข้อมูล เช่น "คอลัมน์ user_id ไม่ควรเป็น null" หรือ "คอลัมน์ age ควรมีค่าระหว่าง 0 ถึง 120"

# build_suite.py
import great_expectations as gx

context = gx.get_context()

# Create or retrieve an Expectation Suite
suite = context.suites.add(
    gx.ExpectationSuite(name="user_events_suite")
)

# Add Expectations to the suite
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+$"
    )
)

# Save the suite to the Data Context
context.suites.save(suite)

GX 1.0+ ใช้ API แบบ class-based สำหรับ Expectations ไวยากรณ์แบบ dictionary เดิม (expect_column_to_exist) ยังคงใช้งานได้ แต่วิธีแบบ class-based ให้การ autocompletion ของ IDE และ type safety ที่ดีกว่า

การรันการตรวจสอบด้วย Checkpoints

Checkpoint เชื่อมโยง Data Asset, Expectation Suite และ Actions ที่เป็นทางเลือกซึ่งถูกเรียกใช้เมื่อการตรวจสอบผ่านหรือล้มเหลว Checkpoints เป็นจุดเข้าหลักสำหรับการตรวจสอบอัตโนมัติใน production

# run_checkpoint.py
import great_expectations as gx

context = gx.get_context()

# Create a 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"),
        ]
    )
)

# Run the Checkpoint
result = checkpoint.run()

# Check overall success
if result.success:
    print("All validations passed")
else:
    print("Validation failures detected")
    for validation_result in result.run_results.values():
        for expectation_result in validation_result.results:
            if not expectation_result.success:
                print(f"  Failed: {expectation_result.expectation_config}")

เมื่อ Checkpoint ทำงาน GX จะโหลด batch ใช้แต่ละ Expectation และอัปเดต Data Docs การตรวจสอบที่ล้มเหลวสามารถเรียกใช้การแจ้งเตือน Slack, alert ของ PagerDuty หรือการหยุด pipeline ขึ้นอยู่กับ Actions ที่กำหนดค่าไว้

การผสาน GX กับ Apache Airflow DAGs

Pipeline ข้อมูล production ส่วนใหญ่ใช้ orchestrator เช่น Apache Airflow GX มีการผสาน Airflow อย่างเป็นทางการ ที่ห่อหุ้มการทำงานของ Checkpoint ใน operator

# 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():
    """Run GX Checkpoint for user events data."""
    context = gx.get_context(context_root_dir="/opt/airflow/gx")
    checkpoint = context.checkpoints.get("user_events_checkpoint")
    result = checkpoint.run()
    
    if not result.success:
        # Raise exception to fail the Airflow task
        raise ValueError("Data validation failed. Check Data Docs for details.")

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
    )
    
    # Downstream tasks depend on validation passing
    # transform_task >> load_task

การวางการตรวจสอบก่อน task transformation จะป้องกันข้อมูลที่ไม่ดีจากการแพร่กระจายไปยัง downstream รูปแบบนี้ บางครั้งเรียกว่า "shift-left testing" จับปัญหาตอน ingestion แทนที่จะรอหลังจากงาน compute ที่มีค่าใช้จ่ายสูงเสร็จสิ้น

คำถามสัมภาษณ์ทั่วไปเกี่ยวกับ Great Expectations

การสัมภาษณ์ data engineering ในปี 2026 มักรวมคำถามเกี่ยวกับเครื่องมือคุณภาพข้อมูล ด้านล่างนี้คือคำถามที่ปรากฏใน technical screen พร้อมกับคำตอบที่แยกแยะผู้สมัครที่มีประสบการณ์จาก junior

"จะใช้งาน data quality checks ใน pipeline production อย่างไร?"

คำตอบที่ดีกล่าวถึงการฝังการตรวจสอบเป็นขั้นตอนของ pipeline ไม่ใช่เป็นแดชบอร์ดแยกต่างหาก โดยเฉพาะ:

• การตรวจสอบ schema ที่ ingestion จับปัญหาโครงสร้างทันที เช่น string ปรากฏในที่ที่คาดหวัง integer
• การตรวจสอบ business logic ตรวจสอบ constraint ของ domain เช่น ราคาที่เป็นบวกและช่วงวันที่
• Alert อัตโนมัติแจ้งวิศวกรเมื่อการตรวจสอบล้มเหลว ป้องกันการเสียหายของข้อมูลอย่างเงียบ ๆ
• dbt tests จัดการการตรวจสอบระดับ transformation ในขณะที่ GX จัดการการตรวจสอบ ingestion และ output

"ความแตกต่างระหว่าง Expectation Suite และ Checkpoint คืออะไร?"

Expectation Suite ประกอบด้วย assertion: คอลัมน์ใดควรมีอยู่ ช่วงค่าใดที่ยอมรับได้ รูปแบบ regex ใดที่ควรตรงกัน มันกำหนด อะไร ที่ต้องตรวจสอบ

Checkpoint กำหนด เมื่อไหร่ และ อย่างไร ที่จะรันการตรวจสอบเหล่านั้น มันเชื่อมต่อ Data Asset เฉพาะ (ข้อมูลที่จะตรวจสอบ), Expectation Suite (กฎที่จะใช้) และ Actions (สิ่งที่เกิดขึ้นหลังการตรวจสอบ)

"จะจัดการ Expectations ที่แตกต่างกันตามสภาพแวดล้อมอย่างไร?"

ข้อมูล production มักมีลักษณะที่แตกต่างจากข้อมูล staging สองวิธีการ:

1. Parameterized Expectations: ใช้ตัวแปรสภาพแวดล้อมหรือพารามิเตอร์ runtime เพื่อปรับ threshold เช่น min_value=int(os.getenv("AGE_MIN", 0))

2. Multiple Suites: รักษา suite แยกต่างหากสำหรับ staging และ production Staging อาจอนุญาต null ใน field ที่เป็นทางเลือกเพื่อทดสอบ data flow ที่ไม่สมบูรณ์

"จะเกิดอะไรขึ้นเมื่อ Checkpoint ล้มเหลวใน pipeline ที่กำหนดเวลาไว้?"

Checkpoint ส่งคืน CheckpointResult ที่มี success=False พฤติกรรม pipeline ขึ้นอยู่กับว่า orchestrator จัดการความล้มเหลวอย่างไร:

• ใน Airflow การ raise exception จะทำเครื่องหมาย task ว่าล้มเหลว บล็อก task downstream
• ใน Databricks notebook สามารถออกด้วยสถานะ error
• Actions ที่แนบกับ Checkpoint สามารถส่งข้อความ Slack สร้าง ticket Jira หรือเรียกใช้ขั้นตอน rollback

Custom Expectations สำหรับการตรวจสอบ Domain-Specific

GX รวม Expectations ในตัวมากกว่า 300 รายการ แต่กฎ domain-specific มักต้องการการ implement แบบ custom Custom Expectation ขยาย base class และ implement logic การตรวจสอบ

# 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):
    """Expect column values to be valid ISO 3166-1 alpha-2 country codes."""
    
    column: str
    
    @classmethod
    def _prescriptive_template(cls) -> str:
        return "Column {column} values must be valid ISO country codes"
    
    def _validate(self, metrics, runtime_configuration=None, execution_engine=None):
        column_values = metrics.get("column_values")
        valid_codes = {c.alpha_2 for c in pycountry.countries}
        
        invalid_values = [
            v for v in column_values 
            if v is not None and v not in valid_codes
        ]
        
        return {
            "success": len(invalid_values) == 0,
            "result": {
                "observed_value": len(invalid_values),
                "unexpected_list": invalid_values[:10]  # Sample
            }
        }

ลงทะเบียน Custom Expectations โดยวางไว้ในไดเรกทอรี great_expectations/plugins/ หรือเพิ่มโมดูลไปที่ plugins_directory ใน great_expectations.yml

Data Docs: การสื่อสารคุณภาพไปยัง Stakeholder

Data Docs เป็นเว็บไซต์ HTML แบบ static ที่สร้างจากผลการตรวจสอบ การรันแต่ละครั้งจะสร้างหน้าที่แสดง Expectations ใดที่ผ่านหรือล้มเหลว พร้อมรายละเอียด drill-down เกี่ยวกับค่าที่ไม่คาดคิด

# Generate and open Data Docs
context = gx.get_context()
context.build_data_docs()
context.open_data_docs()  # Opens browser

สำหรับระบบ production Data Docs สามารถโฮสต์บน S3, GCS หรือ Azure Blob Storage ด้วยการควบคุมการเข้าถึงที่เหมาะสม แพลตฟอร์ม GX Cloud เสนอ Data Docs ที่โฮสต์พร้อมฟีเจอร์การทำงานร่วมกันของทีม

การย้าย GX 1.0: Breaking Changes จาก 0.x

ทีมที่อัปเกรดจาก GX 0.x จะเผชิญกับการเปลี่ยนแปลง API ความแตกต่างหลัก:

คู่มือการย้ายอย่างเป็นทางการ ครอบคลุมแต่ละการเปลี่ยนแปลง สำหรับ codebase ขนาดใหญ่ การย้ายแบบค่อยเป็นค่อยไปโดยใช้ compatibility shims จะลดความเสี่ยง

ประเด็นสำคัญสำหรับคุณภาพข้อมูลด้วย GX 1.22

• Great Expectations 1.22 (สิงหาคม 2026) รองรับ Python 3.10 ถึง 3.13 พร้อมการสนับสนุนทดลอง 3.14 ผ่านตัวแปรสภาพแวดล้อม GX_PYTHON_EXPERIMENTAL
• Expectation Suites กำหนด อะไร ที่จะตรวจสอบ Checkpoints กำหนด เมื่อไหร่ และ อย่างไร
• ฝัง Checkpoints เป็นขั้นตอน pipeline ใน DAGs Airflow หรือ notebooks Databricks เพื่อจับข้อมูลที่ไม่ดีตอน ingestion
• Custom Expectations จัดการกฎ domain-specific ที่ Expectations ในตัวไม่สามารถครอบคลุมได้
• Data Docs ให้รายงาน HTML ที่ stakeholder อ่านได้ โฮสต์บน cloud storage สำหรับการเข้าถึง production
• คำตอบสัมภาษณ์ควรเน้นคุณภาพข้อมูลเป็นขั้นตอน pipeline ที่ฝังอยู่ ไม่ใช่แดชบอร์ดแยกต่างหาก