Data Quality & Testing

# Install Great Expectations
pip install great_expectations

# Initialize project
great_expectations init

# Project structure created:
great_expectations/
├── great_expectations.yml       # Configuration
├── expectations/                # Your test suites
├── checkpoints/                 # Validation configs
├── plugins/                     # Custom expectations
└── uncommitted/                 # Data docs, results

# Connect to data source
import great_expectations as ge

# From pandas DataFrame
df = pd.read_csv('sales_data.csv')
ge_df = ge.from_pandas(df)

# From SQL database
context = ge.get_context()
datasource = context.sources.add_postgres(
    name="my_postgres",
    connection_string="postgresql://localhost/mydb"
)

import great_expectations as ge
import pandas as pd

# Load data as Great Expectations DataFrame
df = pd.read_csv('sales_data.csv')
ge_df = ge.from_pandas(df)

# ============ EXPECTATION 1: Column Existence ============
ge_df.expect_column_to_exist('customer_id')
ge_df.expect_column_to_exist('amount')
ge_df.expect_column_to_exist('order_date')

# ============ EXPECTATION 2: No Nulls ============
ge_df.expect_column_values_to_not_be_null('customer_id')
ge_df.expect_column_values_to_not_be_null('order_id')

# ============ EXPECTATION 3: Unique Values ============
ge_df.expect_column_values_to_be_unique('order_id')

# ============ EXPECTATION 4: Value Ranges ============
ge_df.expect_column_values_to_be_between(
    'amount',
    min_value=0,
    max_value=100000
)

# ============ EXPECTATION 5: Set Membership ============
ge_df.expect_column_values_to_be_in_set(
    'status',
    value_set=['pending', 'completed', 'cancelled', 'refunded']
)

# ============ EXPECTATION 6: Regex Patterns ============
ge_df.expect_column_values_to_match_regex(
    'email',
    regex=r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
)

# ============ EXPECTATION 7: Data Types ============
ge_df.expect_column_values_to_be_of_type('amount', 'float')
ge_df.expect_column_values_to_be_of_type('order_date', 'datetime64')

# ============ EXPECTATION 8: Statistical Properties ============
ge_df.expect_column_mean_to_be_between('amount', min_value=50, max_value=500)
ge_df.expect_column_stdev_to_be_between('amount', min_value=10, max_value=200)

# ============ EXPECTATION 9: Row Count ============
ge_df.expect_table_row_count_to_be_between(min_value=100, max_value=1000000)

# ============ EXPECTATION 10: Referential Integrity ============
# Check foreign key exists in another table
ge_df.expect_column_values_to_be_in_set(
    'customer_id',
    value_set=customers_df['customer_id'].tolist()
)

# Validate all expectations
results = ge_df.validate()

# Check if validation passed
if results['success']:
    print("✓ All expectations passed!")
    print(f"Evaluated {results['statistics']['evaluated_expectations']} expectations")
else:
    print("✗ Validation failed!")
    print(f"Failed: {results['statistics']['unsuccessful_expectations']}")

    # Get details of failures
    for result in results['results']:
        if not result['success']:
            expectation = result['expectation_config']['expectation_type']
            print(f"\nFAILED: {expectation}")
            print(f"Column: {result['expectation_config'].get('kwargs', {}).get('column')}")
            print(f"Details: {result['result']}")

# Example output when validation fails:
"""
✗ Validation failed!
Failed: 3

FAILED: expect_column_values_to_not_be_null
Column: customer_id
Details: {'element_count': 10000, 'unexpected_count': 45,
          'unexpected_percent': 0.45, 'partial_unexpected_list': [None, None, None]}

FAILED: expect_column_values_to_be_between
Column: amount
Details: {'element_count': 10000, 'unexpected_count': 12,
          'unexpected_percent': 0.12, 'unexpected_values': [-50, -100, 150000]}

FAILED: expect_column_mean_to_be_between
Column: amount
Details: {'observed_value': 45.2, 'expected_min': 50, 'expected_max': 500}
"""

# Save validation results
results.to_json_dict()  # Export as JSON
results_df = pd.DataFrame(results['results'])  # Convert to DataFrame

import great_expectations as ge
import pandas as pd

def validate_sales_data(df: pd.DataFrame) -> pd.DataFrame:
    """Validate sales data before loading to warehouse"""

    # Convert to GE DataFrame
    ge_df = ge.from_pandas(df)

    # Define expectations
    ge_df.expect_column_to_exist('customer_id')
    ge_df.expect_column_values_to_not_be_null('customer_id')
    ge_df.expect_column_values_to_be_unique('order_id')
    ge_df.expect_column_values_to_be_between('amount', min_value=0, max_value=100000)
    ge_df.expect_column_values_to_be_in_set(
        'status',
        value_set=['pending', 'completed', 'cancelled']
    )

    # Validate
    results = ge_df.validate()

    if not results['success']:
        # Log failures
        failed_expectations = [
            r for r in results['results'] if not r['success']
        ]

        error_msg = f"Data quality validation failed: {len(failed_expectations)} issues\n"
        for failure in failed_expectations:
            error_msg += f"- {failure['expectation_config']['expectation_type']}\n"
            error_msg += f"  {failure['result']}\n"

        # Raise error to stop pipeline
        raise ValueError(error_msg)

    print(f"✓ Data quality validation passed ({len(df)} rows)")
    return df

# Use in ETL pipeline
def etl_pipeline():
    # Extract
    df = extract_from_source()

    # Validate RAW data (fail fast!)
    df = validate_sales_data(df)

    # Transform (only runs if validation passed)
    df = transform_data(df)

    # Validate TRANSFORMED data
    df = validate_sales_data(df)

    # Load (only runs if both validations passed)
    load_to_warehouse(df)

# In Airflow DAG
from airflow import DAG
from airflow.operators.python import PythonOperator

with DAG('sales_etl', ...) as dag:
    extract_task = PythonOperator(task_id='extract', python_callable=extract_from_source)

    validate_raw = PythonOperator(
        task_id='validate_raw',
        python_callable=validate_sales_data,
        op_kwargs={'df': '{{ ti.xcom_pull(task_ids="extract") }}'}
    )

    transform_task = PythonOperator(task_id='transform', python_callable=transform_data)

    validate_transformed = PythonOperator(
        task_id='validate_transformed',
        python_callable=validate_sales_data
    )

    load_task = PythonOperator(task_id='load', python_callable=load_to_warehouse)

    # Pipeline stops at first validation failure
    extract_task >> validate_raw >> transform_task >> validate_transformed >> load_task

# Build Data Docs (generates HTML documentation)
great_expectations docs build

# Opens in browser:
# - All expectations for each dataset
# - Validation results with pass/fail status
# - Data profiling statistics
# - Trend charts over time

# Data Docs include:
# ✓ Expectation Suite: All validation rules
# ✓ Validation Results: Pass/fail with details
# ✓ Profiling: Statistical summary of data
# ✓ Timeline: Historical validation trends

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

# Initialize Spark
spark = SparkSession.builder \
    .config("spark.jars.packages", "com.amazon.deequ:deequ:2.0.0-spark-3.3") \
    .getOrCreate()

# Load data
df = spark.read.parquet("s3://my-bucket/sales/")

# Define data quality checks
check = Check(spark, CheckLevel.Error, "Sales Data Quality") \
    .hasSize(lambda x: x >= 100) \  # At least 100 rows
    .isComplete("customer_id") \  # No nulls
    .isUnique("order_id") \  # No duplicates
    .isContainedIn("status", ["pending", "completed", "cancelled"]) \
    .isNonNegative("amount") \  # Amount >= 0
    .hasMin("amount", lambda x: x >= 0) \
    .hasMax("amount", lambda x: x <= 100000)

# Run verification
verification_result = VerificationSuite(spark) \
    .onData(df) \
    .addCheck(check) \
    .run()

# Check results
if verification_result.status == "Success":
    print("✓ All data quality checks passed!")
else:
    print("✗ Data quality checks failed!")

    # Get failure details
    result_df = VerificationResult.checkResultsAsDataFrame(spark, verification_result)
    result_df.filter("constraint_status != 'Success'").show()

# Example output (failures):
"""
+-------------------+----------------+------------------+--------------------+
|constraint         |constraint_status|constraint_message|                   |
+-------------------+----------------+------------------+--------------------+
|CompletenessConstraint|Failure      |Value: 0.955     |Expected: 1.0       |
|UniquenessConstraint  |Failure      |Value: 0.998     |Expected: 1.0       |
|ComplianceConstraint  |Failure      |Value: 0.99      |12 values not in set|
+-------------------+----------------+------------------+--------------------+
"""

from pydeequ.analyzers import AnalysisRunner, Completeness, Mean, Size
from pydeequ.repository import FileSystemMetricsRepository, ResultKey

# Setup metrics repository (stores historical metrics)
metrics_repo = FileSystemMetricsRepository(spark, "s3://my-bucket/metrics")

# Define analysis
analysis = AnalysisRunner(spark) \
    .onData(df) \
    .addAnalyzer(Size()) \
    .addAnalyzer(Completeness("customer_id")) \
    .addAnalyzer(Mean("amount")) \
    .useRepository(metrics_repo) \
    .saveOrAppendResult(
        ResultKey(spark, ResultKey.current_milli_time(), tags={"dataset": "sales"})
    ) \
    .run()

# Get current metrics
metrics = analysis.getMetrics()
print(f"Rows: {metrics['Size']['value']}")
print(f"Completeness (customer_id): {metrics['Completeness']['value']}")
print(f"Mean amount: {metrics['Mean']['value']}")

# Anomaly Detection: Compare to historical baseline
from pydeequ.anomaly_detection import SimpleThresholdStrategy, AbsoluteChangeStrategy

# Strategy 1: Simple threshold (alert if below/above threshold)
anomaly_check = Check(spark, CheckLevel.Warning, "Anomaly Detection") \
    .hasSize(lambda x: x >= 1000)  # Alert if < 1000 rows

# Strategy 2: Relative change (alert if deviation from baseline)
# Load historical metrics
historical_metrics = metrics_repo.load() \
    .forAnalyzers([Size(), Mean("amount")]) \
    .getSuccessMetricsAsDataFrame()

# Calculate baseline (e.g., 7-day average)
baseline_size = historical_metrics.filter("name = 'Size'") \
    .selectExpr("avg(value) as baseline").collect()[0]['baseline']
baseline_mean = historical_metrics.filter("name = 'Mean'") \
    .selectExpr("avg(value) as baseline").collect()[0]['baseline']

# Check for anomalies (>20% deviation)
current_size = metrics['Size']['value']
if abs(current_size - baseline_size) / baseline_size > 0.2:
    print(f"⚠️ ANOMALY: Row count changed by {((current_size - baseline_size) / baseline_size) * 100:.1f}%")
    print(f"Expected: {baseline_size}, Got: {current_size}")

current_mean = metrics['Mean']['value']
if abs(current_mean - baseline_mean) / baseline_mean > 0.2:
    print(f"⚠️ ANOMALY: Mean amount changed by {((current_mean - baseline_mean) / baseline_mean) * 100:.1f}%")
    print(f"Expected: {baseline_mean}, Got: {current_mean}")

# AWS Glue Job with Deequ validation
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pydeequ.checks import Check, CheckLevel
from pydeequ.verification import VerificationSuite

# Initialize Glue context
args = getResolvedOptions(sys.argv, ['JOB_NAME'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Read from Glue Data Catalog
datasource = glueContext.create_dynamic_frame.from_catalog(
    database="my_database",
    table_name="sales_data"
)
df = datasource.toDF()

# Define data quality checks
check = Check(spark, CheckLevel.Error, "Glue ETL Quality") \
    .isComplete("customer_id") \
    .isUnique("order_id") \
    .isNonNegative("amount")

# Verify data quality
result = VerificationSuite(spark).onData(df).addCheck(check).run()

if result.status == "Success":
    # Continue ETL if validation passed
    transformed = df.filter("status = 'completed'")

    # Write to S3
    glueContext.write_dynamic_frame.from_options(
        frame=DynamicFrame.fromDF(transformed, glueContext, "transformed"),
        connection_type="s3",
        connection_options={"path": "s3://my-bucket/clean-data/"},
        format="parquet"
    )
else:
    # Fail job if validation failed
    raise Exception("Data quality validation failed!")

import pandas as pd
from pandera import Column, DataFrameSchema, Check

# Define expected schema
sales_schema = DataFrameSchema({
    "order_id": Column(int, nullable=False, unique=True),
    "customer_id": Column(int, nullable=False),
    "amount": Column(float, checks=Check.greater_than_or_equal_to(0)),
    "status": Column(str, checks=Check.isin(['pending', 'completed', 'cancelled'])),
    "order_date": Column(pd.Timestamp, nullable=False),
    "email": Column(str, checks=Check.str_matches(r'^[\w\.-]+@[\w\.-]+\.\w+$'))
})

# Load data
df = pd.read_csv('sales_data.csv')

# Validate schema
try:
    validated_df = sales_schema.validate(df)
    print("✓ Schema validation passed")
except Exception as e:
    print(f"✗ Schema validation failed: {e}")

# Example failures:
"""
CASE 1: Missing column
SchemaError: column 'customer_id' not in dataframe

CASE 2: Wrong data type
SchemaError: expected column 'amount' to have type float64, got object

CASE 3: Constraint violation
SchemaError: <Check greater_than_or_equal_to: 0> failed for column 'amount'
  Values: [-50, -100] violate constraint

CASE 4: Unexpected column
SchemaError: column 'discount' not in schema (strict mode)
"""

# Detect schema drift
def detect_schema_drift(df, expected_schema):
    """Compare current schema to expected"""
    current_cols = set(df.columns)
    expected_cols = set(expected_schema.columns.keys())

    added = current_cols - expected_cols
    removed = expected_cols - current_cols

    if added or removed:
        print("⚠️ Schema drift detected!")
        if added:
            print(f"  New columns: {added}")
        if removed:
            print(f"  Missing columns: {removed}")
        return False

    # Check data types
    for col, expected_type in expected_schema.columns.items():
        actual_type = df[col].dtype
        if actual_type != expected_type.dtype:
            print(f"⚠️ Type mismatch in '{col}': expected {expected_type.dtype}, got {actual_type}")
            return False

    return True

# Schema versioning with Pandera
from pandera import DataFrameSchema, Column, Check
import pandas as pd

# Version 1: Initial schema
schema_v1 = DataFrameSchema({
    "order_id": Column(int),
    "customer_id": Column(int),
    "amount": Column(float)
})

# Version 2: Added 'status' column (backward compatible)
schema_v2 = DataFrameSchema({
    "order_id": Column(int),
    "customer_id": Column(int),
    "amount": Column(float),
    "status": Column(str, nullable=True)  # Nullable for backward compatibility
})

# Version 3: Added 'discount', removed nothing (backward compatible)
schema_v3 = DataFrameSchema({
    "order_id": Column(int),
    "customer_id": Column(int),
    "amount": Column(float),
    "status": Column(str, nullable=True),
    "discount": Column(float, nullable=True, checks=Check.greater_than_or_equal_to(0))
})

# Smart schema validation (try versions newest to oldest)
def validate_with_evolution(df, schemas):
    """Validate against multiple schema versions"""
    for version, schema in sorted(schemas.items(), reverse=True):
        try:
            validated = schema.validate(df, lazy=True)
            print(f"✓ Data matches schema {version}")
            return validated, version
        except Exception as e:
            continue

    raise ValueError("Data doesn't match any known schema version!")

# Usage
schemas = {
    "v1": schema_v1,
    "v2": schema_v2,
    "v3": schema_v3
}

df = pd.read_csv('sales_data.csv')
validated_df, version = validate_with_evolution(df, schemas)
print(f"Data schema version: {version}")

# Migration strategy
if version == "v1":
    # Migrate v1 to v3
    df['status'] = 'pending'  # Add default
    df['discount'] = 0.0      # Add default
    print("Migrated v1 → v3")
elif version == "v2":
    # Migrate v2 to v3
    df['discount'] = 0.0
    print("Migrated v2 → v3")

from ydata_profiling import ProfileReport
import pandas as pd

# Load data
df = pd.read_csv('sales_data.csv')

# Generate comprehensive profile
profile = ProfileReport(
    df,
    title="Sales Data Profile",
    explorative=True,
    correlations={
        "pearson": {"calculate": True},
        "spearman": {"calculate": True},
    }
)

# Export report
profile.to_file("sales_profile.html")

# Profile includes:
# ✓ Dataset statistics (rows, columns, missing %, duplicates %)
# ✓ Variable statistics (type, unique values, missing %)
# ✓ Quantile statistics (min, Q1, median, Q3, max)
# ✓ Descriptive statistics (mean, std, variance, kurtosis, skewness)
# ✓ Histograms and distributions
# ✓ Correlation matrices
# ✓ Missing value patterns
# ✓ Duplicate rows detection
# ✓ Warnings and alerts (high cardinality, high correlation, etc.)

# Example insights from profile:
"""
OVERVIEW:
- 10,000 rows, 15 columns
- 0.5% missing values
- 45 duplicate rows detected

WARNINGS:
- 'customer_email' has 89% unique values (high cardinality)
- 'amount' and 'discount' are highly correlated (0.95)
- 'order_date' has 12 values in the future (data quality issue)
- 'status' has 3 unexpected values not in ['pending', 'completed', 'cancelled']

ANOMALIES:
- 'amount' has 5 outliers (>3 std dev from mean)
- 'quantity' has constant value (0) for 1,234 rows
"""

import pandas as pd
import numpy as np

def detect_anomalies(df, column, method='iqr', threshold=3):
    """Detect anomalies using statistical methods"""

    if method == 'iqr':
        # Interquartile Range (IQR) method
        Q1 = df[column].quantile(0.25)
        Q3 = df[column].quantile(0.75)
        IQR = Q3 - Q1

        lower_bound = Q1 - 1.5 * IQR
        upper_bound = Q3 + 1.5 * IQR

        anomalies = df[(df[column] < lower_bound) | (df[column] > upper_bound)]

    elif method == 'zscore':
        # Z-score method (standard deviations from mean)
        mean = df[column].mean()
        std = df[column].std()

        z_scores = np.abs((df[column] - mean) / std)
        anomalies = df[z_scores > threshold]

    elif method == 'isolation_forest':
        # Machine learning-based (Isolation Forest)
        from sklearn.ensemble import IsolationForest

        model = IsolationForest(contamination=0.1)  # Expect 10% anomalies
        predictions = model.fit_predict(df[[column]].values.reshape(-1, 1))
        anomalies = df[predictions == -1]

    return anomalies

# Apply anomaly detection
df = pd.read_csv('sales_data.csv')

# Detect outliers in 'amount'
amount_anomalies = detect_anomalies(df, 'amount', method='iqr')
print(f"Detected {len(amount_anomalies)} anomalies in 'amount'")
print(f"Anomalous values: {amount_anomalies['amount'].tolist()}")

# Example output:
"""
Detected 23 anomalies in 'amount'
Anomalous values: [150000, 125000, 140000, -50, -100, 180000, ...]

IQR Analysis:
Q1: $45
Q3: $450
IQR: $405
Lower bound: -$162.5 (any negative values are anomalies)
Upper bound: $1,057.5 (amounts > $1,057.5 are anomalies)
"""

# Time-series anomaly detection
def detect_time_series_anomalies(df, timestamp_col, value_col, window='7D'):
    """Detect anomalies in time series using rolling statistics"""

    df = df.sort_values(timestamp_col)
    df[timestamp_col] = pd.to_datetime(df[timestamp_col])
    df = df.set_index(timestamp_col)

    # Calculate rolling mean and std
    rolling_mean = df[value_col].rolling(window=window).mean()
    rolling_std = df[value_col].rolling(window=window).std()

    # Detect anomalies (>3 std from rolling mean)
    upper_bound = rolling_mean + 3 * rolling_std
    lower_bound = rolling_mean - 3 * rolling_std

    anomalies = df[
        (df[value_col] > upper_bound) | (df[value_col] < lower_bound)
    ]

    return anomalies

# Daily sales anomaly detection
daily_sales = df.groupby('order_date')['amount'].sum().reset_index()
sales_anomalies = detect_time_series_anomalies(
    daily_sales,
    'order_date',
    'amount',
    window='7D'
)

print(f"Detected {len(sales_anomalies)} anomalous days")
print(sales_anomalies)

import pytest
import pandas as pd

# Transformation function to test
def clean_sales_data(df):
    """Clean sales data: remove nulls, fix types, filter valid"""
    df = df.dropna(subset=['customer_id', 'order_id'])
    df['amount'] = df['amount'].astype(float).abs()
    df = df[df['amount'] > 0]
    df = df.drop_duplicates(subset=['order_id'])
    return df

# Unit tests
class TestSalesTransformations:

    def test_removes_null_customer_ids(self):
        """Test that rows with null customer_ids are removed"""
        # Arrange
        input_df = pd.DataFrame({
            'customer_id': [1, None, 3],
            'order_id': [101, 102, 103],
            'amount': [100, 200, 300]
        })

        # Act
        result = clean_sales_data(input_df)

        # Assert
        assert len(result) == 2
        assert result['customer_id'].isnull().sum() == 0

    def test_removes_duplicates(self):
        """Test that duplicate order_ids are removed"""
        input_df = pd.DataFrame({
            'customer_id': [1, 1, 2],
            'order_id': [101, 101, 102],  # Duplicate!
            'amount': [100, 100, 200]
        })

        result = clean_sales_data(input_df)

        assert len(result) == 2
        assert result['order_id'].is_unique

    def test_converts_negative_amounts(self):
        """Test that negative amounts are converted to positive"""
        input_df = pd.DataFrame({
            'customer_id': [1, 2],
            'order_id': [101, 102],
            'amount': [-100, -200]
        })

        result = clean_sales_data(input_df)

        assert (result['amount'] > 0).all()
        assert result['amount'].tolist() == [100, 200]

    def test_filters_zero_amounts(self):
        """Test that zero amounts are filtered out"""
        input_df = pd.DataFrame({
            'customer_id': [1, 2, 3],
            'order_id': [101, 102, 103],
            'amount': [100, 0, 300]  # Zero amount
        })

        result = clean_sales_data(input_df)

        assert len(result) == 2
        assert 0 not in result['amount'].values

# Run tests
pytest.main([__file__, '-v'])

# Output:
"""
test_removes_null_customer_ids PASSED
test_removes_duplicates PASSED
test_converts_negative_amounts PASSED
test_filters_zero_amounts PASSED
"""

import pytest
import pandas as pd
from unittest.mock import Mock, patch

# Full pipeline
def etl_pipeline(source_path, dest_path):
    # Extract
    df = pd.read_csv(source_path)

    # Transform
    df = clean_sales_data(df)
    df = aggregate_by_customer(df)

    # Load
    df.to_parquet(dest_path)

    return len(df)

# Integration test
@pytest.fixture
def sample_data_file(tmp_path):
    """Create temporary test data file"""
    data = pd.DataFrame({
        'customer_id': [1, 1, 2, 2, 3],
        'order_id': [101, 102, 103, 104, 105],
        'amount': [100, 150, 200, 50, 300]
    })

    file_path = tmp_path / "test_sales.csv"
    data.to_csv(file_path, index=False)
    return file_path

def test_full_pipeline(sample_data_file, tmp_path):
    """Test complete ETL pipeline end-to-end"""
    # Arrange
    output_path = tmp_path / "output.parquet"

    # Act
    row_count = etl_pipeline(sample_data_file, output_path)

    # Assert
    assert output_path.exists()
    result = pd.read_parquet(output_path)

    # Verify data quality
    assert len(result) == 3  # 3 customers
    assert 'customer_id' in result.columns
    assert result['customer_id'].is_unique
    assert (result['total_amount'] > 0).all()

# Test with mocked database
@patch('pipeline.load_from_database')
def test_pipeline_with_mock_db(mock_load):
    """Test pipeline with mocked database connection"""
    # Mock database response
    mock_load.return_value = pd.DataFrame({
        'customer_id': [1, 2],
        'order_id': [101, 102],
        'amount': [100, 200]
    })

    # Run pipeline
    result = etl_pipeline(source='database', dest='output.parquet')

    # Verify mock was called
    mock_load.assert_called_once()

    # Verify result
    assert result == 2