Hướng dẫn Great Expectations: Xác thực dữ liệu với Python

Thiết lập Great Expectations

Trong hướng dẫn này, bạn sẽ học cách sử dụng GX Core, phiên bản mã nguồn mở của Great Expectations, để xác thực một Pandas DataFrame. Chúng ta sẽ cùng thực hiện thiết lập context, đăng ký nguồn dữ liệu Pandas, định nghĩa expectations và xác thực các lô dữ liệu (batch).

Lưu ý: Chúng tôi khuyến nghị bạn làm theo cùng với notebook DataLab, nhưng bạn cũng có thể tạo script Python của riêng mình.

1. Cài đặt Great Expectations

Yêu cầu tiên quyết

• Đã cài đặt Python 3.9 đến 3.12.
• Để tránh xung đột, rất khuyến khích bạn cài đặt Great Expectations trong môi trường ảo (lưu ý: việc thiết lập môi trường ảo nằm ngoài phạm vi của bài viết này).
• Một tập dữ liệu mẫu.

Lưu ý: Nếu sử dụng notebook DataLab đã cung cấp, các yêu cầu trên đã được đáp ứng. Bạn có thể bỏ qua.

Dùng lệnh sau để cài đặt GX qua pip:

pip install great_expectations

Lệnh này cài đặt gói lõi và mọi phụ thuộc cần thiết.

2. Khởi tạo data context

Great Expectations yêu cầu một data context để quản lý cấu hình. Chúng ta dùng ephemeral data context để tránh lưu cấu hình.

import great_expectations as gx

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

Tạo bộ xác thực dữ liệu đầu tiên của bạn

Sau khi thiết lập GX, hãy tạo một bộ xác thực dữ liệu.

1. Kết nối tới nguồn dữ liệu và tạo data asset

Nguồn dữ liệu kết nối Great Expectations với dữ liệu của bạn, trong khi data asset đại diện cho một phần dữ liệu cụ thể (ví dụ: một bảng, DataFrame hoặc tệp).

Trong trường hợp này, chúng ta sẽ chuẩn bị mọi thứ để kết nối tới một DataFrame tên là inventory_parts_df. Tập dữ liệu mẫu có sẵn trong DataLab đã cung cấp, và sẽ được tạo khi chúng ta chạy khối SQL:

Nếu bạn không dùng DataLab, hãy tạo DataFrame của riêng bạn với dữ liệu mẫu.

Bây giờ, tạo nguồn dữ liệu và asset của bạn:

# 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. Thêm batch definition

Batch definition xác định và tổ chức dữ liệu của bạn để xác thực. Ở đây, chúng ta thêm một batch definition bao phủ toàn bộ 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. Truy xuất một batch

Batch là một tập dữ liệu gắn với batch definition. Để xác thực dữ liệu, bạn sẽ cần truy xuất và liên kết batch với DataFrame của mình, trong trường hợp này là 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. Tạo suite và định nghĩa expectations

Expectations là các quy tắc để xác thực dữ liệu. Trong ví dụ này, chúng ta sẽ định nghĩa các expectation đơn giản sau:

1. Đảm bảo các giá trị inventory_id không được null.
2. Đảm bảo các giá trị part_num là duy nhất.

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

Bạn có thể khám phá tất cả các expectation hiện có trong Expectation Gallery. Hãy thử thêm vài cái nữa!

Sau khi định nghĩa expectations, GX sẽ xuất cấu hình 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
}

Suite bao gồm các chi tiết sau:

1. Tên suite và ID: Tên duy nhất (inventory_parts_suite) và định danh để theo dõi và quản lý suite.
2. Expectations: Mỗi quy tắc chỉ ra:
• Loại kiểm tra (ví dụ: đảm bảo một cột không có giá trị null hoặc có các giá trị duy nhất).
• Các tham số, như cột đang được xác thực.
• Metadata và ID duy nhất cho mỗi expectation, giúp dễ theo dõi và tùy chỉnh.
3. Metadata: Thông tin phiên bản của Great Expectations, đảm bảo tương thích với công cụ.
4. Ghi chú: Nơi để thêm nhận xét mô tả về suite (tùy chọn).

Đầu ra có cấu trúc này vừa là tài liệu, vừa là cấu hình có thể tái sử dụng để xác thực tập dữ liệu của bạn, giúp các expectation được định nghĩa rõ ràng, có thể truy vết và sẵn sàng cho lần dùng sau.

5. Xác thực dữ liệu

Cuối cùng, xác thực batch theo các expectation đã định nghĩa và đánh giá kết quả.

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

Sau khi chạy xác thực, Great Expectations cung cấp báo cáo chi tiết về việc liệu tập dữ liệu có đáp ứng các expectation đã định nghĩa hay không:

{
  "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
}

Báo cáo này chi tiết chất lượng dữ liệu của bạn, làm nổi bật phần đạt và phần chưa đạt. Dưới đây là giải thích đơn giản về kết quả:

Xác thực tổng thể: Kết quả xác thực thành công một phần: 50% expectation đạt và 50% không đạt. Một expectation không đạt cho thấy có vấn đề chất lượng dữ liệu cần được xử lý. Trong trường hợp này, một cột không đáp ứng quy tắc đã đặt.

Expectation 1: inventory_id không được có giá trị thiếu

• Kết quả: Đạt
• Giải thích: Mọi giá trị trong cột inventory_id đều có mặt, không có bản ghi null hay thiếu. Điều này cho thấy độ đầy đủ dữ liệu tốt cho cột này.

Expectation 2: part_num phải có các giá trị duy nhất

• Kết quả: Không đạt
• Giải thích: Cột part_num chứa 97,98% giá trị trùng lặp, nghĩa là chỉ có một vài giá trị là duy nhất.
• Điểm nổi bật:
• Ví dụ các giá trị trùng gồm "3069b" và "33291".
• Công cụ cũng hiển thị tần suất xuất hiện của các bản sao này và vị trí hàng của chúng, giúp dễ dàng định vị và khắc phục vấn đề.

Dĩ nhiên, đây chỉ là một tập dữ liệu mẫu, và chúng tôi chủ đích bao gồm một expectation đạt và một expectation trượt để bạn thấy cả hai kết quả xác thực.

Vậy là xong! Bạn đã chạy xác thực dữ liệu end-to-end thành công.

Tích hợp Great Expectations vào các pipeline dữ liệu

Trong môi trường sản xuất, các bước xác thực phải được nhúng trực tiếp vào quy trình để liên tục giám sát chất lượng dữ liệu ở mọi giai đoạn. 

Trong phần này, chúng ta sẽ thảo luận cách bạn có thể tích hợp Great Expectations vào các pipeline dữ liệu của mình.

Đây là các ví dụ để bạn hình dung, và có thể cần thêm cấu hình không nêu ở đây. Hãy xem tài liệu của từng công cụ để có cú pháp mới nhất!

Tích hợp với công cụ ETL

Tích hợp Great Expectations với các công cụ ETL phổ biến như Apache Airflow hoặc Prefect tương đối đơn giản. Nhúng các bước xác thực trực tiếp vào quy trình ETL sẽ giúp bạn phát hiện và xử lý vấn đề dữ liệu theo thời gian thực trước khi chúng ảnh hưởng đến phân tích phía sau.

Hãy cùng đi qua một ví dụ đơn giản tích hợp Great Expectations với Prefect để chạy xác thực dữ liệu như một phần của quy trình ETL tự động:

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

Trong ví dụ này, chúng ta định nghĩa một flow Prefect với một task để chạy xác thực bằng Great Expectations.

Hàm validate_data() tải context của Great Expectations, truy xuất batch dữ liệu và áp dụng expectation suite. 

Nếu dữ liệu không đáp ứng tiêu chí xác thực, task sẽ đưa ra cảnh báo, dừng workflow và ngăn lỗi lan sang các bước sau.

Xác thực dữ liệu liên tục

Bạn có thể lập lịch chạy xác thực bằng nhiều công cụ, như cron trên hệ Unix hoặc các dịch vụ quản lý như Apache Airflow. Trong ví dụ này, chúng ta sẽ minh họa cách lập lịch chạy xác thực bằng Airflow, vốn rất phù hợp để điều phối các pipeline dữ liệu.

Dưới đây là cách bạn có thể thiết lập một DAG (Đồ thị Có hướng Không chu trình) của Airflow để chạy xác thực Great Expectations hằng ngày:

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

Trong ví dụ này, chúng ta định nghĩa một DAG lập lịch chạy xác thực mỗi ngày (@daily). 

Hàm run_validation() thực thi xác thực bằng cách tải context của Great Expectations và chạy expectation suite đã định nghĩa trên dữ liệu.

Thực hành tốt nhất khi xác thực dữ liệu với Great Expectations

Tuân theo các thực hành tốt nhất luôn được khuyến nghị để đảm bảo khả năng mở rộng và hiệu quả, và điều này cũng đúng với xác thực dữ liệu bằng Great Expectations.

Bắt đầu nhỏ và lặp lại

Bắt đầu với các kiểm tra chất lượng dữ liệu nền tảng và mở rộng dần. Tốt hơn hết là tập trung vào các expectation cơ bản lúc đầu để tránh làm phức tạp quá trình, giúp tích hợp mượt mà hơn và dễ gỡ lỗi hơn. Khi hiểu rõ hơn về tập dữ liệu, bạn có thể thêm các xác thực phức tạp hơn.

Phối hợp giữa các nhóm

Chất lượng dữ liệu không chỉ là mối quan tâm kỹ thuật. Hãy phối hợp với các nhóm nghiệp vụ để định nghĩa expectations và đảm bảo xác thực được triển khai phù hợp với logic và mục tiêu kinh doanh cốt lõi. Cách tiếp cận liên chức năng này đảm bảo dữ liệu phục vụ đúng mục đích và đáp ứng yêu cầu của mọi bên liên quan.

Tự động hóa khi có thể

Tự động hóa quy trình ở mức tối đa khả thi để tích hợp xác thực dữ liệu vào các pipeline. Tích hợp các kiểm tra xác thực tự động cho phép giám sát liên tục chất lượng dữ liệu mà không cần can thiệp thủ công, từ đó cải thiện đáng kể hiệu suất.

Kết luận

Làm tốt lắm! Bạn đã học cách cấu hình và xác thực dữ liệu trong Great Expectations. Những kỹ thuật này sẽ giúp duy trì chất lượng dữ liệu cao và tính minh bạch trong quy trình làm việc của bạn.

Để tiếp tục nâng cao kỹ năng, hãy tham khảo các tài nguyên sau:

• ETL và ELT trong Python: Học cách biến đổi và di chuyển dữ liệu hiệu quả.
• Giới thiệu về Chất lượng Dữ liệu: Khám phá các nguyên lý cơ bản của quản lý chất lượng dữ liệu.
• Làm sạch Dữ liệu trong Python: Thành thạo các kỹ thuật làm sạch dữ liệu để đảm bảo độ chính xác và nhất quán.
• Cheat Sheet về Các chiều chất lượng dữ liệu: Tài liệu tham khảo nhanh về các chiều chất lượng dữ liệu.