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ẽ đi qua việc 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 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 Python 3.9 đến 3.12.
• Để tránh xung đột, rất khuyến nghị bạn cài Great Expectations trong một môi trường ảo (lưu ý: thiết lập môi trường ảo không nằm trong phạm vi bài viết này).
• Một tập dữ liệu mẫu.

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

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

pip install great_expectations

Lệnh này cài 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, còn data asset đại diện cho một tập con cụ thể của dữ liệu (ví dụ: 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ờ, hãy tạo nguồn dữ liệu và 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. Thêm batch definition

Batch definition nhận diện 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 hợp dữ liệu gắn với batch definition. Để xác thực dữ liệu, bạn 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 giá trị inventory_id không null.
2. Đảm bảo 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. Chúng tôi khuyến khích bạn thêm một vài cái nữa!

Sau khi định nghĩa expectations, GX 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 và ID của suite: Một 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ỉ rõ:
• Loại kiểm tra (ví dụ: đảm bảo một cột không có giá trị null hoặc có giá trị duy nhất).
• Các tham số, như cột được xác thực.
• Siêu dữ liệu và một ID duy nhất cho mỗi expectation, giúp việc theo dõi và tùy chỉnh dễ dàng hơn.
3. Metadata: Thông tin phiên bản của Great Expectations, đảm bảo tính tương thích với công cụ.
4. Ghi chú: Nơi để thêm bình luận 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, nhờ đó các expectation được định nghĩa rõ ràng, có thể truy vết và sẵn sàng cho lần sử dụng sau.

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

Cuối cùng, xác thực batch dựa trên 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 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 mô tả chất lượng dữ liệu của bạn, nêu bật các mục đạt và không đạ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% thất bại. Một expectation thất bại cho thấy 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 đã định nghĩa.

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

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

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

• Kết quả: Thất bại
• Giải thích: Cột part_num chứa 97,98% giá trị trùng lặp, nghĩa là chỉ có rất ít giá trị là duy nhất.
• Điểm nổi bật:
• Ví dụ giá trị trùng lặp 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 trùng và vị trí dòng của chúng, giúp bạn dễ dàng định vị và khắc phục vấn đề.

Dĩ nhiên, đây chỉ là tập dữ liệu mẫu, và chúng tôi cố ý đưa vào 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 production, các bước xác thực cần đượ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, có thể cần thêm cấu hình không bao gồm ở đây. Hãy xem tài liệu của từng công cụ để cập nhật cú pháp mới nhất!

Tích hợp với các 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 ảnh hưởng đến phân tích hạ nguồn.

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 workflow 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 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ẽ phát cảnh báo, dừng workflow và ngăn lỗi lan xuống hạ nguồn.

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

Bạn có thể lập lịch các job xác thực bằng nhiều công cụ, như cron job trên hệ thống 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ên lịch chạy xác thực mỗi ngày (@daily). 

Hàm run_validation() thực thi việc xác thực bằng cách tải context của Great Expectations và chạy expectation suite đã định nghĩa với 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ở rộng và hiệu quả, và điều đó cũng đúng với việc xác thực dữ liệu bằng Great Expectations.

Bắt đầu nhỏ và lặp dần

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 ban đầ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. Khi bạn hiểu rõ hơn về tập dữ liệu, có thể bổ sung các xác thực phức tạp hơn.

Hợp tác liên phòng ban

Chất lượng dữ liệu không chỉ là mối quan tâm kỹ thuật. Hãy hợp tác với các nhóm nghiệp vụ để định nghĩa expectation và đảm bảo xác thực triển khai phù hợp với logic và mục tiêu kinh doanh. 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 có thể để 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 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 quả.

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 chuyể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 kiến thức nền tảng về quản lý chất lượng dữ liệu.
• Làm sạch Dữ liệu trong Python: Thành thạo 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 tiện dụng về các chiều chất lượng dữ liệu.