第08章 数据质量¶

import great_expectations as gx
import pandas as pd

# 创建 Data Context
context = gx.get_context()

# 读取数据并创建 Data Source
df = pd.read_csv('sales.csv')
data_source = context.data_sources.add_pandas(name="sales_source")
data_asset = data_source.add_dataframe_asset(name="sales_asset")
batch_definition = data_asset.add_batch_definition_whole_dataframe("sales_batch")
batch = batch_definition.get_batch(batch_parameters={"dataframe": df})

# 创建 Expectation Suite
suite = context.suites.add(
    gx.ExpectationSuite(name="sales_suite")
)

# 添加 Expectations
suite.add_expectation(
    gx.expectations.ExpectColumnToExist(column="product_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="quantity", min_value=1, max_value=1000
    )
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="amount", min_value=0, max_value=100000
    )
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="customer_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="sale_id")
)

# 创建 Validation Definition 并验证
validation_definition = context.validation_definitions.add(
    gx.ValidationDefinition(
        name="sales_validation",
        data=batch_definition,
        suite=suite
    )
)
result = validation_definition.run(batch_parameters={"dataframe": df})

# 输出结果
print(f"Validation success: {result.success}")
for exp_result in result.results:
    print(f"  {exp_result.expectation_config.type}: {exp_result.success}")

from pydeequ.checks import Check, CheckLevel
from pydeequ.verification import VerificationSuite, VerificationResult
from pyspark.sql import SparkSession

# 创建SparkSession
spark = SparkSession.builder \
    .appName("DeequExample") \
    .config("spark.jars.packages", "com.amazon.deequ:deequ:2.0.7-spark-3.5") \
    .getOrCreate()

# 读取数据
df = spark.read.json('sales.json')

# 创建检查规则
check = Check(spark, CheckLevel.Error, "sales_data_check") \
    .isComplete("product_id") \
    .isComplete("product_name") \
    .isUnique("sale_id") \
    .isNonNegative("quantity") \
    .isNonNegative("amount") \
    .satisfies("quantity >= 1 AND quantity <= 1000", "quantity_range") \
    .satisfies("amount >= 0 AND amount <= 100000", "amount_range")

# 运行验证
result = VerificationSuite(spark) \
    .onData(df) \
    .addCheck(check) \
    .run()

# 输出结果
result_df = VerificationResult.checkResultsAsDataFrame(spark, result)
result_df.show(truncate=False)
print(f"Verification status: {result.status}")

class DataQualityMonitor:
    """数据质量监控类"""

    def __init__(self):
        self.metrics = {}

    def check_completeness(self, df):
        """检查完整性"""
        metrics = {}

        # 缺失值检查
        for column in df.columns:
            missing_count = df[column].isnull().sum()
            missing_ratio = missing_count / len(df)
            metrics[f'{column}_missing_ratio'] = missing_ratio

        # 重复检查
        duplicate_count = df.duplicated().sum()
        metrics['duplicate_ratio'] = duplicate_count / len(df)

        return metrics

    def check_accuracy(self, df):
        """检查准确性"""
        metrics = {}

        # 格式检查
        if 'date' in df.columns:
            try:  # try/except捕获异常
                pd.to_datetime(df['date'])
                metrics['date_format_valid'] = True
            except:
                metrics['date_format_valid'] = False

        # 值域检查
        if 'quantity' in df.columns:
            invalid_count = df[(df['quantity'] < 0) | (df['quantity'] > 1000)].shape[0]
            metrics['quantity_valid_ratio'] = 1 - invalid_count / len(df)

        return metrics

    def check_consistency(self, df):
        """检查一致性"""
        metrics = {}

        # 逻辑一致性
        if 'amount' in df.columns and 'quantity' in df.columns and 'price' in df.columns:
            calculated_amount = df['quantity'] * df['price']
            consistency = (df['amount'] - calculated_amount).abs() < 0.01
            metrics['amount_consistency_ratio'] = consistency.mean()

        return metrics

    def run_checks(self, df):
        """运行所有检查"""
        completeness = self.check_completeness(df)
        accuracy = self.check_accuracy(df)
        consistency = self.check_consistency(df)

        self.metrics = {
            'completeness': completeness,
            'accuracy': accuracy,
            'consistency': consistency
        }

        return self.metrics

    def alert(self, threshold=0.05):
        """告警"""
        for category, metrics in self.metrics.items():
            for metric_name, value in metrics.items():
                if value < threshold:
                    print(f"ALERT: {category}.{metric_name} = {value}")