data-quality-checker（数据质量检查器）

Data Quality Checker

Implement comprehensive data quality checks and validation.

Quick Start

Use Great Expectations for validation, implement schema checks, monitor data quality metrics, set up alerts.

Instructions

Great Expectations Setup

import great_expectations as gx

context = gx.get_context()

# Create expectation suite
suite = context.add_expectation_suite("data_quality_suite")

# Add expectations
validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite_name="data_quality_suite"
)

# Schema validation
validator.expect_table_columns_to_match_ordered_list(
    column_list=["id", "name", "email", "created_at"]
)

# Null checks
validator.expect_column_values_to_not_be_null("email")

# Value ranges
validator.expect_column_values_to_be_between("age", min_value=0, max_value=120)

# Uniqueness
validator.expect_column_values_to_be_unique("email")

# Run validation
results = validator.validate()

Custom Validation Rules

def validate_data_quality(df):
    issues = []

    # Check for nulls
    null_counts = df.isnull().sum()
    if null_counts.any():
        issues.append(f"Null values found: {null_counts[null_counts > 0]}")

    # Check for duplicates
    duplicates = df.duplicated().sum()
    if duplicates > 0:
        issues.append(f"Found {duplicates} duplicate rows")

    # Check data freshness
    max_date = df['created_at'].max()
    if (datetime.now() - max_date).days > 1:
        issues.append("Data is stale")

    return issues

Data Quality Metrics

def calculate_quality_metrics(df):
    return {
        'completeness': 1 - (df.isnull().sum().sum() / df.size),
        'uniqueness': df.drop_duplicates().shape[0] / df.shape[0],
        'validity': (df['email'].str.contains('@').sum() / len(df)),
        'timeliness': (datetime.now() - df['created_at'].max()).days
    }

Best Practices

• Validate at ingestion
• Monitor quality metrics
• Set up alerts for failures
• Document quality rules
• Regular quality audits
• Track quality trends