บทเรียน Great Expectations: ตรวจสอบความถูกต้องของข้อมูลด้วย Python

การตั้งค่า Great Expectations

ในบทเรียนนี้ จะได้เรียนรู้การใช้ GX Core เวอร์ชันโอเพ่นซอร์สของ Great Expectations เพื่อตรวจสอบความถูกต้องของ Pandas DataFrame เราจะพาดูการตั้งค่า context การลงทะเบียนแหล่งข้อมูล Pandas การกำหนด expectations และการตรวจสอบชุดข้อมูล (batch)

หมายเหตุ: แนะนำให้ทำตามไปพร้อมกับโน้ตบุ๊ก DataLab แต่คุณสามารถสร้างสคริปต์ Python ของตนเองได้เช่นกัน

1. การติดตั้ง Great Expectations

ข้อกำหนดเบื้องต้น

• ติดตั้ง Python 3.9 ถึง 3.12 แล้ว
• เพื่อหลีกเลี่ยงความขัดแย้ง ขอแนะนำอย่างยิ่งให้ติดตั้ง Great Expectations ภายใน virtual environment (ข้อจำกัดความรับผิดชอบ: การตั้งค่า virtual environment อยู่นอกขอบเขตของบทความนี้)
• ชุดข้อมูลตัวอย่าง

หมายเหตุ: หากใช้โน้ตบุ๊ก DataLab ที่ให้ไว้ ข้อกำหนดข้างต้นถูกเตรียมไว้เรียบร้อยแล้ว สามารถข้ามขั้นตอนนี้ได้

ใช้คำสั่งต่อไปนี้เพื่อติดตั้ง GX ผ่าน pip:

pip install great_expectations

คำสั่งนี้จะติดตั้งแพ็กเกจหลักและไลบรารีที่จำเป็นทั้งหมด

2. เริ่มต้นใช้งาน data context

Great Expectations ต้องใช้ data context เพื่อจัดการคอนฟิก เราจะใช้ ephemeral data context เพื่อหลีกเลี่ยงการบันทึกคอนฟิกไว้ถาวร

import great_expectations as gx

# Get the Ephemeral Data Context
context = gx.get_context()
assert type(context).__name__ == "EphemeralDataContext"

สร้างชุดการตรวจสอบข้อมูลครั้งแรกของคุณ

เมื่อเตรียม GX เรียบร้อยแล้ว มาสร้างชุดการตรวจสอบข้อมูลกัน

1. เชื่อมต่อแหล่งข้อมูลและสร้าง data asset

Data source เชื่อม Great Expectations กับข้อมูลของคุณ ขณะที่ data asset แทนชุดข้อมูลเฉพาะ (เช่น ตาราง DataFrame หรือไฟล์)

ในที่นี้ เราจะเตรียมทุกอย่างเพื่อเชื่อมต่อกับ DataFrame ที่ชื่อ inventory_parts_df ชุดข้อมูลตัวอย่างมีให้ใน DataLab ที่จัดเตรียมไว้ และจะถูกสร้างขึ้นเมื่อรันบล็อก SQL:

หากไม่ได้ใช้ DataLab ให้สร้าง DataFrame ของตนเองพร้อมข้อมูลตัวอย่าง

ต่อไป สร้าง data source และ asset ของคุณ:

# Add a Pandas Data Source
data_source = context.data_sources.add_pandas(name="inventory_parts")
# Add a Data Asset to the Data Source
data_asset = data_source.add_dataframe_asset(name="inventory_parts_asset")

2. เพิ่ม batch definition

Batch definition ใช้ระบุและจัดระเบียบข้อมูลสำหรับการตรวจสอบ ที่นี่ เราจะเพิ่ม batch definition ที่ครอบคลุมทั้ง DataFrame:

# Define the Batch Definition name
batch_definition_name = "inventory_parts_batch"
# Add the Batch Definition
batch_definition = data_asset.add_batch_definition_whole_dataframe(batch_definition_name)
assert batch_definition.name == batch_definition_name

3. ดึงข้อมูล batch

Batch คือชุดข้อมูลที่ผูกกับ batch definition ในการตรวจสอบข้อมูล ต้องดึง batch และเชื่อมกับ DataFrame ของคุณ ซึ่งในที่นี้คือ inventory_parts_df:

# Define the Batch Parameters
batch_parameters = {"dataframe": inventory_parts_df}
# Retrieve the Batch
batch = batch_definition.get_batch(batch_parameters=batch_parameters)

4. สร้างชุดและกำหนด expectations

Expectations คือกฎสำหรับตรวจสอบความถูกต้องของข้อมูล ในตัวอย่างนี้ เราจะกำหนด expectations แบบง่าย 2 ข้อดังนี้:

1. ตรวจให้แน่ใจว่า inventory_id ไม่มีค่าเป็นค่าว่าง (non-null)
2. ตรวจให้แน่ใจว่า part_num มีค่าไม่ซ้ำกัน (unique)

# Create an Expectation Suite
expectation_suite_name = "inventory_parts_suite"
suite = gx.ExpectationSuite(name=expectation_suite_name)
# Add Expectations
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="inventory_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="part_num")
)
# Add the Expectation Suite to the Context
context.suites.add(suite)

สามารถสำรวจ expectations ทั้งหมดได้ใน Expectation Gallery ขอแนะนำให้ลองเพิ่มอีกสักสองสามข้อ

หลังจากกำหนด expectations แล้ว GX จะส่งออกคอนฟิกของ expectation suite ดังนี้:

{
  "name": "inventory_parts_suite",
  "id": "b2de0b69-0869-4163-8dde-6c09884483f7",
  "expectations": [
    {
      "type": "expect_column_values_to_not_be_null",
      "kwargs": {
        "column": "inventory_id"
      },
      "meta": {},
      "id": "53d6c42a-d190-412f-a113-783b706531f4"
    },
    {
      "type": "expect_column_values_to_be_unique",
      "kwargs": {
        "column": "part_num"
      },
      "meta": {},
      "id": "362a2bdc-616d-4b3a-b7f0-c73808caee78"
    }
  ],
  "meta": {
    "great_expectations_version": "1.2.4"
  },
  "notes": null
}

ภายในชุดนี้ประกอบด้วยรายละเอียดต่อไปนี้:

1. ชื่อและ ID ของชุด: ชื่อเฉพาะ (inventory_parts_suite) และตัวระบุที่ใช้ติดตามและจัดการชุด
2. Expectations: แต่ละกฎระบุ:
• ประเภทของการตรวจสอบ (เช่น ตรวจว่าคอลัมน์ไม่มีค่า null หรือมีค่าไม่ซ้ำ)
• พารามิเตอร์ เช่น คอลัมน์ที่ใช้ตรวจสอบ
• เมทาดาทาและ ID เฉพาะของแต่ละ expectation เพื่อให้ง่ายต่อการติดตามและปรับแต่ง
3. เมทาดาทา: ข้อมูลเวอร์ชันของ Great Expectations เพื่อความเข้ากันได้ของเครื่องมือ
4. บันทึก: ช่องสำหรับเพิ่มคำอธิบายชุด (ไม่บังคับ)

เอาต์พุตที่มีโครงสร้างนี้ทำหน้าที่เป็นทั้งเอกสารและคอนฟิกที่นำกลับมาใช้ซ้ำได้สำหรับการตรวจสอบชุดข้อมูลของคุณ ทำให้ expectations ถูกกำหนดไว้อย่างชัดเจน ตรวจสอบย้อนกลับได้ และพร้อมใช้งานในอนาคต

5. ตรวจสอบความถูกต้องของข้อมูล

สุดท้าย ตรวจสอบ batch เทียบกับ expectations ที่กำหนดไว้และประเมินผลลัพธ์

# Validate the Data Against the Suite
validation_results = batch.validate(suite)
# Evaluate the Results
print(validation_results)

หลังจากรันการตรวจสอบแล้ว Great Expectations จะให้รายงานรายละเอียดว่าชุดข้อมูลเป็นไปตาม expectations ที่กำหนดไว้หรือไม่:

{
  "success": false,
  "results": [
    {
      "success": true,
      "expectation_config": {
        "type": "expect_column_values_to_not_be_null",
        "kwargs": {
          "batch_id": "inventory_parts-inventory_parts_asset",
          "column": "inventory_id"
        },
        "meta": {},
        "id": "53d6c42a-d190-412f-a113-783b706531f4"
      },
      "result": {
        "element_count": 580069,
        "unexpected_count": 0,
        "unexpected_percent": 0.0,
        "partial_unexpected_list": [],
        "partial_unexpected_counts": [],
        "partial_unexpected_index_list": []
      },
      "meta": {},
      "exception_info": {
        "raised_exception": false,
        "exception_traceback": null,
        "exception_message": null
      }
    },
    {
      "success": false,
      "expectation_config": {
        "type": "expect_column_values_to_be_unique",
        "kwargs": {
          "batch_id": "inventory_parts-inventory_parts_asset",
          "column": "part_num"
        },
        "meta": {},
        "id": "362a2bdc-616d-4b3a-b7f0-c73808caee78"
      },
      "result": {
        "element_count": 580069,
        "unexpected_count": 568352,
        "unexpected_percent": 97.98006788847535,
        "partial_unexpected_list": [
          "48379c01",
          "paddle",
          "11816pr0005",
          "2343",
          "3003",
          "30176",
          "3020",
          "3022",
          "3023",
          "30357",
          "3039",
          "3062b",
          "3068b",
          "3069b",
          "3069b",
          "33291",
          "33291",
          "3795",
          "3941",
          "3960"
        ],
        "missing_count": 0,
        "missing_percent": 0.0,
        "unexpected_percent_total": 97.98006788847535,
        "unexpected_percent_nonmissing": 97.98006788847535,
        "partial_unexpected_counts": [
          {
            "value": "3069b",
            "count": 2
          },
          {
            "value": "33291",
            "count": 2
          },
          {
            "value": "11816pr0005",
            "count": 1
          },
          {
            "value": "2343",
            "count": 1
          },
          {
            "value": "3003",
            "count": 1
          },
          {
            "value": "30176",
            "count": 1
          },
          {
            "value": "3020",
            "count": 1
          },
          {
            "value": "3022",
            "count": 1
          },
          {
            "value": "3023",
            "count": 1
          },
          {
            "value": "30357",
            "count": 1
          },
          {
            "value": "3039",
            "count": 1
          },
          {
            "value": "3062b",
            "count": 1
          },
          {
            "value": "3068b",
            "count": 1
          },
          {
            "value": "3795",
            "count": 1
          },
          {
            "value": "3941",
            "count": 1
          },
          {
            "value": "3960",
            "count": 1
          },
          {
            "value": "48379c01",
            "count": 1
          },
          {
            "value": "paddle",
            "count": 1
          }
        ],
        "partial_unexpected_index_list": [
          0,
          3,
          4,
          5,
          6,
          7,
          8,
          9,
          10,
          11,
          12,
          13,
          14,
          15,
          16,
          17,
          18,
          19,
          20,
          21
        ]
      },
      "meta": {},
      "exception_info": {
        "raised_exception": false,
        "exception_traceback": null,
        "exception_message": null
      }
    }
  ],
  "suite_name": "inventory_parts_suite",
  "suite_parameters": {},
  "statistics": {
    "evaluated_expectations": 2,
    "successful_expectations": 1,
    "unsuccessful_expectations": 1,
    "success_percent": 50.0
  },
  "meta": {
    "great_expectations_version": "1.2.4",
    "batch_spec": {
      "batch_data": "PandasDataFrame"
    },
    "batch_markers": {
      "ge_load_time": "20241129T122532.416424Z",
      "pandas_data_fingerprint": "84a1e1939091fcf54324910def3b89cd"
    },
    "active_batch_definition": {
      "datasource_name": "inventory_parts",
      "data_connector_name": "fluent",
      "data_asset_name": "inventory_parts_asset",
      "batch_identifiers": {
        "dataframe": "<DATAFRAME>"
      }
    }
  },
  "id": null
}

รายงานนี้แจกแจงคุณภาพของข้อมูล โดยไฮไลต์ทั้งส่วนที่ผ่านและไม่ผ่าน ต่อไปนี้คือคำอธิบายผลลัพธ์แบบย่อ:

ภาพรวมการตรวจสอบ: ผลการตรวจสอบสำเร็จบางส่วน: ผ่าน 50% และไม่ผ่าน 50% การไม่ผ่านหมายถึงมีปัญหาคุณภาพข้อมูลที่ต้องแก้ไข ในกรณีนี้มีคอลัมน์หนึ่งไม่เป็นไปตามกฎที่กำหนด

Expectation 1: inventory_id ต้องไม่มีค่าที่หายไป

• ผลลัพธ์: ผ่าน
• คำอธิบาย: ทุกค่าภายในคอลัมน์ inventory_id มีข้อมูลครบถ้วน ไม่มีค่า null หรือข้อมูลที่หายไป แสดงถึงความครบถ้วนของข้อมูลในคอลัมน์นี้

Expectation 2: part_num ต้องมีค่าไม่ซ้ำกัน

• ผลลัพธ์: ไม่ผ่าน
• คำอธิบาย: คอลัมน์ part_num มีค่าซ้ำ 97.98% หมายความว่ามีค่าส่วนน้อยเท่านั้นที่ไม่ซ้ำ
• ไฮไลต์:
• ตัวอย่างค่าที่ซ้ำ เช่น "3069b" และ "33291"
• เครื่องมือยังแสดงความถี่ของค่าที่ซ้ำและตำแหน่งแถว ทำให้ง่ายต่อการระบุและแก้ไขปัญหา

แน่นอนว่านี่เป็นเพียงชุดข้อมูลตัวอย่าง และเราได้ตั้งใจใส่ทั้งกรณีที่ผ่านและไม่ผ่านไว้ เพื่อให้เห็นผลการตรวจสอบทั้งสองแบบ

เพียงเท่านี้ คุณก็ได้รันการตรวจสอบข้อมูลแบบ end-to-end สำเร็จแล้ว

การผสาน Great Expectations เข้ากับดาต้าไปป์ไลน์

ในสภาพแวดล้อมการใช้งานจริง การตรวจสอบต้องฝังอยู่ในเวิร์กโฟลว์โดยตรง เพื่อมอนิเตอร์คุณภาพข้อมูลอย่างต่อเนื่องในทุกขั้นตอน

ในส่วนนี้ เราจะพูดถึงวิธีผสาน Great Expectations เข้ากับดาต้าไปป์ไลน์ของคุณ

ตัวอย่างเหล่านี้มีไว้เพื่อให้เห็นภาพเท่านั้น อาจต้องมีคอนฟิกเพิ่มเติมที่ไม่ได้ระบุไว้ที่นี่ โปรดดูเอกสารของแต่ละเครื่องมือเพื่อไวยากรณ์ล่าสุด

การผสานกับเครื่องมือ ETL

การผสาน Great Expectations กับเครื่องมือ ETL ยอดนิยมอย่าง Apache Airflow หรือ Prefect ทำได้ค่อนข้างตรงไปตรงมา การฝังขั้นตอนการตรวจสอบไว้ในกระบวนการ ETL โดยตรงจะช่วยให้ตรวจจับและแก้ไขปัญหาข้อมูลได้แบบเรียลไทม์ ก่อนที่จะส่งผลกระทบต่อการวิเคราะห์ปลายน้ำ

มาดูตัวอย่างง่ายๆ ของการผสาน Great Expectations กับ Prefect เพื่อรันการตรวจสอบข้อมูลเป็นส่วนหนึ่งของเวิร์กโฟลว์ ETL อัตโนมัติ:

from prefect import task, Flow
import great_expectations as ge
# Define a task to run Great Expectations validation
@task
def validate_data():
    context = ge.data_context.DataContext()
    batch_kwargs = {"path": "path/to/your/datafile.csv", "datasource": "your_datasource"}
    batch = context.get_batch(batch_kwargs, suite_name="your_expectation_suite")
    results = context.run_validation_operator("action_list_operator", assets_to_validate=[batch])
	    
    # Check validation results and raise an alert if validation fails
    if not results["success"]:
        raise ValueError("Data validation failed!")
# Define your ETL flow
with Flow("ETL_with_GE_Validation") as flow:
    validation = validate_data()
# Execute the flow
flow.run()

ในตัวอย่างนี้ เรากำหนด Prefect flow พร้อมงานสำหรับรันการตรวจสอบด้วย Great Expectations

ฟังก์ชัน validate_data() โหลด context ของ Great Expectations ดึง batch ของข้อมูล และใช้ expectation suite

หากข้อมูลไม่เป็นไปตามเกณฑ์การตรวจสอบ งานจะยกการแจ้งเตือนเพื่อหยุดเวิร์กโฟลว์ ป้องกันข้อผิดพลาดที่ปลายน้ำ

การตรวจสอบข้อมูลอย่างต่อเนื่อง

สามารถตั้งเวลาให้รันงานตรวจสอบได้ด้วยเครื่องมือหลากหลาย เช่น cron บนระบบ Unix หรือบริการที่มีการจัดการอย่าง Apache Airflow สำหรับตัวอย่างนี้ เราจะแสดงวิธีตั้งเวลาการรันด้วย Airflow ซึ่งเหมาะสำหรับการ orchestrate ดาต้าไปป์ไลน์

ต่อไปนี้คือวิธีตั้งค่า Airflow DAG (Directed Acyclic Graph) เพื่อรันการตรวจสอบ Great Expectations รายวัน:

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
import great_expectations as ge
# Define the DAG and set the schedule to run daily
default_args = {
	  'owner': 'airflow',
	  'start_date': datetime(2024, 1, 1),
	  'retries': 1,
}
dag = DAG(
      'great_expectations_validation',
	default_args=default_args,
	schedule_interval='@daily',  # Runs once a day
)
# Define the function to run the validation
def run_validation():
    context = ge.data_context.DataContext()
    batch = context.get_batch(batch_kwargs, suite_name="your_expectation_suite")
    results = context.run_validation_operator("action_list_operator", assets_to_validate=[batch])
    return results
# Set up the task in Airflow
validation_task = PythonOperator(
      task_id='run_great_expectations_validation',
      python_callable=run_validation,
      dag=dag,
)
# Set the task in the DAG
validation_task

ตัวอย่างนี้กำหนด DAG ที่ตั้งเวลาให้รันการตรวจสอบวันละครั้ง (@daily)

ฟังก์ชัน run_validation() จะดำเนินการตรวจสอบโดยโหลด context ของ Great Expectations และรัน expectation suite ที่กำหนดกับข้อมูล

แนวทางปฏิบัติที่ดีสำหรับการตรวจสอบข้อมูลด้วย Great Expectations

การยึดแนวทางปฏิบัติที่ดีช่วยเรื่องการขยายและประสิทธิภาพ ซึ่งการตรวจสอบข้อมูลด้วย Great Expectations ก็เช่นเดียวกัน

เริ่มจากสิ่งเล็กและปรับเพิ่มเป็นลำดับ

เริ่มจากการตรวจสอบคุณภาพข้อมูลพื้นฐานและค่อยๆ ขยายต่อไป การโฟกัสที่ expectations พื้นฐานในระยะแรกช่วยหลีกเลี่ยงความซับซ้อนเกินจำเป็น ทำให้บูรณาการได้ราบรื่นและแก้ปัญหาได้ง่ายขึ้น เมื่อเข้าใจชุดข้อมูลมากขึ้น ค่อยเพิ่มการตรวจสอบที่ซับซ้อนขึ้น

ทำงานร่วมกันข้ามทีม

คุณภาพข้อมูลไม่ใช่แค่ประเด็นทางเทคนิค ควรร่วมมือกับทีมธุรกิจในการกำหนด expectations เพื่อให้การตรวจสอบที่นำไปใช้สอดคล้องกับตรรกะและเป้าหมายทางธุรกิจ วิธีการทำงานร่วมกันข้ามสายงานจะช่วยให้ข้อมูลตอบวัตถุประสงค์และตรงตามความต้องการของผู้มีส่วนได้ส่วนเสียทุกฝ่าย

ทำให้เป็นอัตโนมัติเมื่อเป็นไปได้

ทำให้กระบวนการเป็นอัตโนมัติเท่าที่ทำได้ เพื่อนำการตรวจสอบข้อมูลเข้าไปในดาต้าไปป์ไลน์ การตรวจสอบแบบอัตโนมัติทำให้มอนิเตอร์คุณภาพข้อมูลได้อย่างต่อเนื่องโดยไม่ต้องพึ่งพางานด้วยมือ ซึ่งช่วยเพิ่มประสิทธิภาพได้อย่างมาก

สรุป

เยี่ยมมาก คุณได้เรียนรู้การคอนฟิกและตรวจสอบความถูกต้องของข้อมูลใน Great Expectations แล้ว เทคนิคเหล่านี้ช่วยรักษาคุณภาพข้อมูลและความโปร่งใสในเวิร์กโฟลว์ของคุณ

หากต้องการต่อยอดทักษะ ลองดูแหล่งข้อมูลเหล่านี้:

• ETL and ELT in Python: เรียนรู้การแปลงและย้ายข้อมูลอย่างมีประสิทธิภาพ
• Introduction to Data Quality: ทำความเข้าใจกับพื้นฐานการจัดการคุณภาพข้อมูล
• Cleaning Data in Python: เชี่ยวชาญเทคนิคการทำความสะอาดข้อมูลเพื่อความถูกต้องและความสม่ำเสมอ
• Data Quality Dimensions Cheat Sheet: คู่มือสรุปมิติคุณภาพข้อมูลที่ใช้งานสะดวก