Thiết kế Pipeline Kiểm tra Tính hợp lệ & Làm sạch Dữ liệu Khai báo

Mục tiêu kiến thức

• Xây dựng động cơ kiểm soát chất lượng dữ liệu (Data Quality & Schema Validation Engine) theo hướng khai báo (Declarative).
• Thực thi các quy tắc kiểm thử nghiêm ngặt: Kiểm tra sự tồn tại của cột, giá trị rỗng (nullable), kiểm tra kiểu (int, float, string), giới hạn miền giá trị (min, max), và biểu thức chính quy (regex).
• Phát hiện và loại bỏ các bản ghi trùng lặp khóa chính (dedup_keys), chỉ giữ lại bản ghi hợp lệ xuất hiện đầu tiên.
• Phân tách luồng dữ liệu thành hai bảng: Bảng dữ liệu sạch đã ép kiểu chuẩn (valid_df) và Bảng dữ liệu lỗi kèm danh sách lý do vi phạm (invalid_df).

Mô tả bài toán

Trong các hệ thống Data Lakehouse hiện đại, dữ liệu thô (Raw Data) từ nhiều nguồn thứ ba gửi về thường chứa lỗi: thiếu trường, sai định dạng email/SĐT, số âm vô lý, hoặc trùng lặp mã đơn hàng. Để đảm bảo dữ liệu đưa vào kho không bị ô nhiễm, cần có một pipeline kiểm duyệt tự động.

Hãy viết hàm: validate_and_clean_data(df: pd.DataFrame, schema: dict) -> tuple[pd.DataFrame, pd.DataFrame]

Cấu trúc đối tượng schema:

schema = {
    "columns": {
        "col_name": {
            "type": "int" | "float" | "string",
            "nullable": bool,
            "min": float | int,    # Tùy chọn (chỉ áp dụng cho int/float)
            "max": float | int,    # Tùy chọn (chỉ áp dụng cho int/float)
            "regex": str           # Tùy chọn (chỉ áp dụng cho string)
        }
    },
    "dedup_keys": list[str]        # Tùy chọn danh sách cột khóa duy nhất
}

Quy tắc kiểm tra từng dòng theo thứ tự ưu tiên:

• Kiểm tra cột bắt buộc:
• Nếu một cột khai báo trong schema["columns"] hoàn toàn không tồn tại trong df.columns, TOÀN BỘ các dòng của df đều bị coi là lỗi với mã lý do: f"MISSING_COLUMN_{col}".
• Kiểm tra Giá trị Rỗng (nullable):
• Nếu ô dữ liệu là rỗng (pd.isna(val)) mà cột đó có nullable=False: thêm lý do f"NULL_NOT_ALLOWED_{col}".
• Kiểm tra Kiểu & Giới hạn (với ô không rỗng):
• Thử ép kiểu:
• Với type == 'int': Thử chuyển sang số nguyên (chuỗi biểu diễn số nguyên hợp lệ hoặc số thực có phần thập phân bằng 0). Nếu thất bại, ghi nhận f"INVALID_TYPE_{col}".
• Với type == 'float': Thử chuyển sang số thực float(val). Nếu thất bại, ghi nhận f"INVALID_TYPE_{col}".
• Với type == 'string': Giá trị phải là chuỗi ký tự (hoặc ép sang chuỗi).
• Kiểm tra miền giá trị (nếu ép kiểu thành công cho int/float):
• Nếu có khai báo min và val < min: ghi nhận f"OUT_OF_RANGE_MIN_{col}".
• Nếu có khai báo max và val > max: ghi nhận f"OUT_OF_RANGE_MAX_{col}".
• Kiểm tra định dạng Regex (cho string):
• Nếu có khai báo regex và re.match(regex, str(val)) không khớp: ghi nhận f"PATTERN_MISMATCH_{col}".
• Kiểm tra Trùng lặp Khóa (dedup_keys):
• Chỉ áp dụng kiểm tra trùng lặp cho các dòng không vi phạm bất kỳ quy tắc nào ở trên.
• Nếu giá trị tổ hợp các cột trong dedup_keys đã từng xuất hiện ở một dòng hợp lệ trước đó, dòng hiện tại bị đánh dấu lỗi với lý do "DUPLICATE_KEY".
• Đầu ra:
• valid_df: DataFrame gồm các dòng hợp lệ, các cột được ép kiểu chuẩn mực (int64, float64, object), reset index drop=True.
• invalid_df: DataFrame gồm các dòng vi phạm, giữ nguyên giá trị ban đầu, thêm cột mới error_reasons (dạng list[str]), reset index drop=True.
• Trả về (valid_df, invalid_df).

Ràng buộc

• Số dòng dữ liệu: 0 ≤ N ≤ 10⁴.
• Schema luôn có khóa "columns" hợp lệ.