Công Cụ Làm Sạch Dữ Liệu & Thực Hành Tốt Nhất: OpenRefine, Thư Viện Python và Giải Pháp Tự Động Hóa

[
  {
    "op": "core/text-transform",
    "description": "Text transform on cells in column country using expression value.trim().toLowercase()",
    "engineConfig": {"mode": "row-based", "facets": []},
    "columnName": "country",
    "expression": "value.trim().toLowercase()",
    "onError": "keep-original"
  }
]

import pandas as pd
import numpy as np

# Load dữ liệu
df = pd.read_csv('messy_data.csv')

# 1. Xử lý missing values
# Đếm missing values
df.isnull().sum()

# Drop rows có missing values (ít hơn 5%)
df = df.dropna(thresh=len(df.columns) * 0.95)

# Fill missing values
df['age'] = df['age'].fillna(df['age'].median())  # Numeric: median
df['category'] = df['category'].fillna('Unknown')  # Categorical: mode hoặc 'Unknown'
df['temperature'] = df['temperature'].interpolate(method='linear')  # Time series: interpolate

# 2. Remove duplicates
df = df.drop_duplicates()
df = df.drop_duplicates(subset=['name', 'email'], keep='first')

# 3. String operations và regex
df['email'] = df['email'].str.lower().str.strip()
df['phone'] = df['phone'].str.replace(r'[^0-9]', '', regex=True)
df['name'] = df['name'].str.title()  # "john doe" → "John Doe"

# 4. Type conversion
df['date'] = pd.to_datetime(df['date'], errors='coerce', format='%Y-%m-%d')
df['price'] = pd.to_numeric(df['price'], errors='coerce')
df['category'] = df['category'].astype('category')  # Tiết kiệm bộ nhớ

# 5. Outlier detection với IQR
Q1 = df['revenue'].quantile(0.25)
Q3 = df['revenue'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
outliers = df[(df['revenue'] < lower_bound) | (df['revenue'] > upper_bound)]

# 6. Z-score method
from scipy import stats
z_scores = np.abs(stats.zscore(df['revenue']))
df_clean = df[z_scores < 3]  # Giữ lại giá trị có z-score < 3

# 7. Categorical encoding
df['category_encoded'] = df['category'].map({'A': 0, 'B': 1, 'C': 2})

import janitor

# Thay vì nhiều dòng code thủ công
df = (df
    .clean_names()  # Chuyển tên cột thành snake_case
    .remove_empty()  # Xóa rows/columns hoàn toàn empty
    .dropna(subset=['critical_column'])
    .filter_on('status == "active"')
    .transform_column('revenue', np.log1p, 'log_revenue')
    .encode_categorical(['category', 'region'])
    .remove_columns(['temp_column'])
)

import datatest

def validate_data(df):
    # Kiểm tra schema
    datatest.validate(df.columns, {'id', 'name', 'date', 'revenue'})
    
    # Kiểm tra kiểu dữ liệu
    datatest.validate(df['id'], int)
    datatest.validate(df['date'], pd.Timestamp)
    
    # Kiểm tra ràng buộc
    datatest.validate(df['revenue'], lambda x: x >= 0)
    datatest.validate(df['status'], {'active', 'inactive', 'pending'})

from cleanlab.classification import CleanLearning
from sklearn.ensemble import RandomForestClassifier

# Phát hiện label errors
cl = CleanLearning(RandomForestClassifier(), verbose=True)
label_issues = cl.find_label_issues(X, y)

# Xem các bản ghi nghi ngờ bị gán nhãn sai
suspicious = label_issues[label_issues['is_label_issue'] == True]
print(suspicious[['given_label', 'predicted_label', 'label_quality_score']])

from autoclean import AutoClean

# Làm sạch tự động
df_clean = AutoClean(df, mode='auto').output

# Mode options:
# - 'auto': Tự động phát hiện và xử lý
# - 'manual': Cho phép tùy chỉnh từng bước

import klib

# Phân tích dữ liệu
klib.cat_plot(df)  # Visualization cho categorical data
klib.corr_plot(df)  # Correlation matrix
klib.dist_plot(df)  # Distribution plots

# Làm sạch dữ liệu
df_clean = klib.data_cleaning(df)
df_clean = klib.convert_datatypes(df)
df_clean = klib.drop_missing(df)

import great_expectations as gx

# Khởi tạo context
context = gx.get_context()

# Kết nối data source
datasource = context.sources.add_pandas("my_datasource")
data_asset = datasource.add_dataframe_asset(name="sales_data")

# Định nghĩa expectations
validator = context.get_validator(
    datasource_name="my_datasource",
    data_asset_name="sales_data",
    batch_request=data_asset.build_batch_request(dataframe=df)
)

# Thêm expectations
validator.expect_column_values_to_not_be_null("customer_id")
validator.expect_column_values_to_be_between("age", min_value=0, max_value=120)
validator.expect_column_values_to_be_in_set("status", ["active", "inactive"])
validator.expect_column_mean_to_be_between("revenue", 1000, 50000)
validator.expect_column_values_to_match_regex("email", r"^.+@.+\\..+")

# Lưu expectation suite
validator.save_expectation_suite(discard_failed_expectations=False)

# Chạy checkpoint
checkpoint = context.add_checkpoint(
    name="my_checkpoint",
    validations=[{
        "batch_request": data_asset.build_batch_request(dataframe=df),
        "expectation_suite_name": "my_expectation_suite"
    }]
)
checkpoint_result = checkpoint.run()

# Little's MCAR test (thư viện missingno hoặc pymc)
import missingno as msno

# Visualize missing patterns
msno.matrix(df)  # Hiển thị pattern missing
msno.heatmap(df)  # Correlation giữa các cột missing

# Iterative imputation (MAR)
from sklearn.impute import IterativeImputer
imputer = IterativeImputer(random_state=42)
df_imputed = pd.DataFrame(imputer.fit_transform(df), columns=df.columns)

# Exact duplicates
df = df.drop_duplicates()

# Fuzzy matching với thefuzz
from thefuzz import fuzz, process

def find_fuzzy_duplicates(df, column, threshold=85):
    values = df[column].unique()
    duplicates = []
    for i, val in enumerate(values):
        matches = process.extract(val, values[i+1:], scorer=fuzz.ratio, limit=3)
        for match, score, _ in matches:
            if score >= threshold:
                duplicates.append((val, match, score))
    return duplicates

# Record linkage với recordlinkage
import recordlinkage
indexer = recordlinkage.Index()
indexer.block('last_name')  # Block trước để giảm số cặp so sánh
pairs = indexer.index(df_a, df_b)

compare = recordlinkage.Compare()
compare.exact('first_name', 'first_name', label='first_name')
compare.string('address', 'address', method='jarowinkler', label='address')
features = compare.compute(pairs, df_a, df_b)

# Isolation Forest — phù hợp datasets lớn
from sklearn.ensemble import IsolationForest

iso = IsolationForest(contamination=0.01, random_state=42, n_jobs=-1)
outliers = iso.fit_predict(df[['feature1', 'feature2']])
df['is_outlier'] = outliers == -1

# Local Outlier Factor
from sklearn.neighbors import LocalOutlierFactor
lof = LocalOutlierFactor(n_neighbors=20, contamination=0.01, n_jobs=-1)
outliers = lof.fit_predict(df[['feature1', 'feature2']])

# DBSCAN cho spatial outliers
from sklearn.cluster import DBSCAN
dbscan = DBSCAN(eps=0.5, min_samples=5)
clusters = dbscan.fit_predict(df[['feature1', 'feature2']])
df['is_outlier'] = clusters == -1

from great_expectations.core import ExpectationSuite

def detect_schema_drift(current_df, expected_schema):
    """Phát hiện schema drift"""
    issues = []
    
    # Kiểm tra cột mất
    missing_cols = set(expected_schema['columns']) - set(current_df.columns)
    if missing_cols:
        issues.append(f"Missing columns: {missing_cols}")
    
    # Kiểm tra cột mới
    new_cols = set(current_df.columns) - set(expected_schema['columns'])
    if new_cols:
        issues.append(f"New columns: {new_cols}")
    
    # Kiểm tra type changes
    for col, expected_type in expected_schema['dtypes'].items():
        if col in current_df.columns:
            actual_type = current_df[col].dtype
            if actual_type != expected_type:
                issues.append(f"Type drift in {col}: {expected_type} → {actual_type}")
    
    return issues

# Sử dụng logging
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def clean_data(df):
    original_shape = df.shape
    logger.info(f"Original data shape: {original_shape}")
    
    df = df.drop_duplicates()
    logger.info(f"After dedup: {df.shape} (removed {original_shape[0] - df.shape[0]} rows)")
    
    missing_before = df.isnull().sum().sum()
    df = df.fillna({'age': df['age'].median()})
    logger.info(f"Filled {missing_before} missing values")
    
    return df

# Sử dụng Makefile hoặc pipeline framework
# hoặc đơn giản là một Python script với requirements.txt
"""
# clean_data.py
# Dependencies: pandas==2.0.0, numpy==1.24.0

Usage:
    python clean_data.py --input raw.csv --output clean.csv --config config.yaml
"""

# Sử dụng Git để version các scripts làm sạch
git add src/cleaning/
git commit -m "Add data cleaning pipeline v2 - handle new date format"

def validate_cleaned_data(df):
    """Kiểm tra dữ liệu sau khi làm sạch"""
    assert df.isnull().sum().sum() == 0, "Còn missing values!"
    assert df.duplicated().sum() == 0, "Còn duplicates!"
    assert df['revenue'].min() >= 0, "Revenue không âm!"
    assert df['date'].max() <= pd.Timestamp.now(), "Không có ngày tương lai!"
    print("Validation passed!")

# Cấu trúc thư mục
# data/
#   raw/          # Dữ liệu gốc, không bao giờ sửa
#   interim/      # Dữ liệu trung gian
#   processed/    # Dữ liệu cuối cùng

import pandas as pd
from ydata_profiling import ProfileReport

# Tạo báo cáo chất lượng dữ liệu
df = pd.read_csv('data.csv')
profile = ProfileReport(df, title="Data Quality Report")
profile.to_file("data_quality_report.html")

# data_contract.yaml
table: user_transactions
version: 2.0
schema:
  transaction_id:
    type: string
    required: true
    unique: true
  amount:
    type: float
    required: true
    min: 0
    max: 1000000
  timestamp:
    type: datetime
    required: true
    format: ISO-8601

# pipeline/clean.py
import pandas as pd
import great_expectations as gx
import logging
from pathlib import Path

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class DataCleaningPipeline:
    def __init__(self, config):
        self.config = config
        self.expectations = []
    
    def load(self, filepath):
        logger.info(f"Loading {filepath}")
        return pd.read_csv(filepath)
    
    def profile(self, df):
        logger.info(f"Shape: {df.shape}, Memory: {df.memory_usage().sum() / 1e6:.2f} MB")
        logger.info(f"Missing: {df.isnull().sum().sum()}, Duplicates: {df.duplicated().sum()}")
        return df
    
    def clean(self, df):
        # Remove duplicates
        df = df.drop_duplicates()
        
        # Handle missing values theo config
        for col, strategy in self.config.get('missing_strategy', {}).items():
            if strategy == 'median':
                df[col] = df[col].fillna(df[col].median())
            elif strategy == 'mode':
                df[col] = df[col].fillna(df[col].mode()[0])
            elif strategy == 'drop':
                df = df.dropna(subset=[col])
        
        # Type conversion
        for col, dtype in self.config.get('dtypes', {}).items():
            if dtype == 'datetime':
                df[col] = pd.to_datetime(df[col], errors='coerce')
            else:
                df[col] = df[col].astype(dtype)
        
        return df
    
    def validate(self, df):
        for expectation in self.config.get('expectations', []):
            # Run Great Expectations validations
            pass
        return True
    
    def run(self, input_path, output_path):
        df = self.load(input_path)
        df = self.profile(df)
        df = self.clean(df)
        self.validate(df)
        df.to_csv(output_path, index=False)
        logger.info(f"Exported to {output_path}")
        return df

# Usage
pipeline = DataCleaningPipeline(config={
    'missing_strategy': {'age': 'median', 'name': 'drop'},
    'dtypes': {'date': 'datetime', 'amount': 'float'}
})
pipeline.run('data/raw/sales.csv', 'data/processed/sales_clean.csv')

# .github/workflows/data_quality.yml
name: Data Quality Check
on: [push, pull_request]
jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - run: pip install -r requirements.txt
      - run: python -m pytest tests/test_data_quality.py
      - run: python pipeline/clean.py --validate-only

# Thêm missing indicator
df['age_missing'] = df['age'].isnull().astype(int)
df['age'] = df['age'].fillna(df['age'].median())

# Thêm vào scripts
import hashlib

def log_checksum(filepath):
    """Log MD5 checksum để xác minh dữ liệu"""
    with open(filepath, 'rb') as f:
        checksum = hashlib.md5(f.read()).hexdigest()
    logger.info(f"File {filepath} MD5: {checksum}")
    return checksum