undefined

**Accuracy**: Does data match expected patterns?
```python

**Consistency**: Are values within expected ranges?
```python

**Uniqueness**: Are keys unique?
```python

**Timeliness**: Is data fresh?
```python

undefined

import great_expectations as gx
from great_expectations.core.batch import RuntimeBatchRequest

**Expectation Suite:**
```python

undefined

import dlt
from pyspark.sql.functions import *

@dlt.table(
    name="validated_orders",
    comment="Orders with comprehensive quality validation"
)

undefined

from typing import Dict, Any, List
from pyspark.sql import DataFrame
from pyspark.sql.functions import *

class BusinessRuleValidator:
    """Custom validator for complex business rules."""

    def __init__(self, spark_session):
        self.spark = spark_session
        self.validation_results = []

    def validate_order_logic(self, df: DataFrame) -> Dict[str, Any]:
        """Validate complex order business rules."""

        # Rule 1: Total matches sum of line items
        total_mismatch = df.filter(
            abs(col("order_total") - col("line_items_sum")) > 0.01
        ).count()

        # Rule 2: Discount doesn't exceed maximum allowed
        invalid_discount = df.filter(
            col("discount_amount") > col("order_subtotal") * 0.5
        ).count()

        # Rule 3: Shipping date after order date
        invalid_dates = df.filter(
            col("shipping_date") < col("order_date")
        ).count()

        return {
            "total_records": df.count(),
            "total_mismatch_count": total_mismatch,
            "invalid_discount_count": invalid_discount,
            "invalid_dates_count": invalid_dates,
            "pass_rate": (
                df.count() - total_mismatch - invalid_discount - invalid_dates
            ) / df.count() * 100
        }

    def validate_referential_integrity(
        self,
        df: DataFrame,
        reference_table: str,
        key_column: str
    ) -> Dict[str, Any]:
        """Check referential integrity constraints."""

        ref_df = self.spark.table(reference_table)

        # Find orphaned records
        orphaned = df.join(
            ref_df,
            df[key_column] == ref_df[key_column],
            "left_anti"
        )

        orphaned_count = orphaned.count()

        return {
            "total_records": df.count(),
            "orphaned_records": orphaned_count,
            "orphaned_percentage": orphaned_count / df.count() * 100,
            "valid": orphaned_count == 0
        }

"""
Great Expectations suite for customer data validation.
"""
import great_expectations as gx
from great_expectations.core.batch import RuntimeBatchRequest
from pyspark.sql import SparkSession
from typing import Dict, Any


class CustomerDataQualitySuite:
    """Comprehensive quality suite for customer data."""

    def __init__(self, context_root_dir: str = None):
        self.context = gx.get_context(context_root_dir=context_root_dir)
        self.suite_name = "customer_data_quality"

    def create_suite(self) -> gx.core.ExpectationSuite:
        """Create comprehensive expectation suite."""

        suite = self.context.create_expectation_suite(
            expectation_suite_name=self.suite_name,
            overwrite_existing=True
        )

        # Schema validation
        suite.add_expectation(
            gx.core.ExpectationConfiguration(
                expectation_type="expect_table_columns_to_match_ordered_list",
                kwargs={
                    "column_list": [
                        "customer_id",
                        "first_name",
                        "last_name",
                        "email",
                        "phone",
                        "registration_date",
                        "status",
                        "lifetime_value"
                    ]
                }
            )
        )

        # Completeness checks
        for col in ["customer_id", "email", "registration_date"]:
            suite.add_expectation(
                gx.core.ExpectationConfiguration(
                    expectation_type="expect_column_values_to_not_be_null",
                    kwargs={"column": col}
                )
            )

        # Uniqueness
        suite.add_expectation(
            gx.core.ExpectationConfiguration(
                expectation_type="expect_column_values_to_be_unique",
                kwargs={"column": "customer_id"}
            )
        )

        suite.add_expectation(
            gx.core.ExpectationConfiguration(
                expectation_type="expect_column_values_to_be_unique",
                kwargs={"column": "email"}
            )
        )

        # Format validation
        suite.add_expectation(
            gx.core.ExpectationConfiguration(
                expectation_type="expect_column_values_to_match_regex",
                kwargs={
                    "column": "email",
                    "regex": r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}$"
                }
            )
        )

        suite.add_expectation(
            gx.core.ExpectationConfiguration(
                expectation_type="expect_column_values_to_match_regex",
                kwargs={
                    "column": "phone",
                    "regex": r"^\+?[1-9]\d{1,14}$",
                    "mostly": 0.95  # Allow 5% invalid
                }
            )
        )

        # Value constraints
        suite.add_expectation(
            gx.core.ExpectationConfiguration(
                expectation_type="expect_column_values_to_be_in_set",
                kwargs={
                    "column": "status",
                    "value_set": ["active", "inactive", "suspended", "closed"]
                }
            )
        )

        suite.add_expectation(
            gx.core.ExpectationConfiguration(
                expectation_type="expect_column_values_to_be_between",
                kwargs={
                    "column": "lifetime_value",
                    "min_value": 0,
                    "max_value": 1000000,
                    "mostly": 0.99
                }
            )
        )

        # Date validation
        suite.add_expectation(
            gx.core.ExpectationConfiguration(
                expectation_type="expect_column_values_to_be_between",
                kwargs={
                    "column": "registration_date",
                    "min_value": "2020-01-01",
                    "max_value": "2026-12-31",
                    "parse_strings_as_datetimes": True
                }
            )
        )

        # Row count check
        suite.add_expectation(
            gx.core.ExpectationConfiguration(
                expectation_type="expect_table_row_count_to_be_between",
                kwargs={
                    "min_value": 100,
                    "max_value": 10000000
                }
            )
        )

        self.context.save_expectation_suite(suite)
        return suite

    def run_validation(self, df: DataFrame) -> Dict[str, Any]:
        """Run validation against dataframe."""

        # Create runtime batch request
        batch_request = RuntimeBatchRequest(
            datasource_name="spark_datasource",
            data_connector_name="runtime_connector",
            data_asset_name="customer_data",
            runtime_parameters={"batch_data": df},
            batch_identifiers={"default_identifier_name": "default_identifier"}
        )

        # Run checkpoint
        checkpoint_result = self.context.run_checkpoint(
            checkpoint_name="customer_data_checkpoint",
            validations=[
                {
                    "batch_request": batch_request,
                    "expectation_suite_name": self.suite_name
                }
            ]
        )

        return {
            "success": checkpoint_result.success,
            "statistics": checkpoint_result.run_results,
            "validation_results": checkpoint_result.to_json_dict()
        }

"""
Custom data quality framework for comprehensive validation.
"""
from typing import Dict, List, Any, Callable, Optional
from pyspark.sql import DataFrame, SparkSession
from pyspark.sql.functions import *
from dataclasses import dataclass
from datetime import datetime
import json


@dataclass
class QualityRule:
    """Define a quality rule."""
    name: str
    description: str
    severity: str  # 'critical', 'high', 'medium', 'low'
    validation_func: Callable[[DataFrame], int]
    threshold: float = 1.0  # Pass rate threshold


@dataclass
class QualityResult:
    """Quality validation result."""
    rule_name: str
    passed: bool
    total_records: int
    failed_records: int
    pass_rate: float
    severity: str
    timestamp: datetime
    details: Dict[str, Any]


class DataQualityFramework:
    """Comprehensive data quality validation framework."""

    def __init__(self, spark: SparkSession):
        self.spark = spark
        self.rules: List[QualityRule] = []
        self.results: List[QualityResult] = []

    def add_rule(self, rule: QualityRule) -> None:
        """Add validation rule to framework."""
        self.rules.append(rule)

    def add_completeness_rule(
        self,
        column: str,
        severity: str = "high"
    ) -> None:
        """Add completeness check for a column."""

        def check(df: DataFrame) -> int:
            return df.filter(col(column).isNull()).count()

        self.add_rule(QualityRule(
            name=f"completeness_{column}",
            description=f"Check {column} has no null values",
            severity=severity,
            validation_func=check
        ))

    def add_uniqueness_rule(
        self,
        column: str,
        severity: str = "critical"
    ) -> None:
        """Add uniqueness check for a column."""

        def check(df: DataFrame) -> int:
            total = df.count()
            unique = df.select(column).distinct().count()
            return total - unique

        self.add_rule(QualityRule(
            name=f"uniqueness_{column}",
            description=f"Check {column} values are unique",
            severity=severity,
            validation_func=check
        ))

    def add_format_rule(
        self,
        column: str,
        regex_pattern: str,
        severity: str = "medium"
    ) -> None:
        """Add format validation rule."""

        def check(df: DataFrame) -> int:
            return df.filter(
                ~col(column).rlike(regex_pattern)
            ).count()

        self.add_rule(QualityRule(
            name=f"format_{column}",
            description=f"Check {column} matches pattern {regex_pattern}",
            severity=severity,
            validation_func=check
        ))

    def add_range_rule(
        self,
        column: str,
        min_value: Any,
        max_value: Any,
        severity: str = "medium"
    ) -> None:
        """Add range validation rule."""

        def check(df: DataFrame) -> int:
            return df.filter(
                (col(column) < min_value) | (col(column) > max_value)
            ).count()

        self.add_rule(QualityRule(
            name=f"range_{column}",
            description=f"Check {column} is between {min_value} and {max_value}",
            severity=severity,
            validation_func=check
        ))

    def add_referential_integrity_rule(
        self,
        column: str,
        reference_table: str,
        reference_column: str,
        severity: str = "critical"
    ) -> None:
        """Add referential integrity check."""

        def check(df: DataFrame) -> int:
            ref_df = self.spark.table(reference_table)
            orphaned = df.join(
                ref_df,
                df[column] == ref_df[reference_column],
                "left_anti"
            )
            return orphaned.count()

        self.add_rule(QualityRule(
            name=f"ref_integrity_{column}",
            description=f"Check {column} exists in {reference_table}.{reference_column}",
            severity=severity,
            validation_func=check
        ))

    def add_custom_rule(
        self,
        name: str,
        description: str,
        validation_func: Callable[[DataFrame], int],
        severity: str = "medium",
        threshold: float = 1.0
    ) -> None:
        """Add custom validation rule."""

        self.add_rule(QualityRule(
            name=name,
            description=description,
            severity=severity,
            validation_func=validation_func,
            threshold=threshold
        ))

    def validate(self, df: DataFrame) -> Dict[str, Any]:
        """Run all validation rules."""

        self.results = []
        total_records = df.count()

        for rule in self.rules:
            try:
                failed_count = rule.validation_func(df)
                pass_rate = (total_records - failed_count) / total_records * 100

                result = QualityResult(
                    rule_name=rule.name,
                    passed=pass_rate >= (rule.threshold * 100),
                    total_records=total_records,
                    failed_records=failed_count,
                    pass_rate=pass_rate,
                    severity=rule.severity,
                    timestamp=datetime.now(),
                    details={"description": rule.description}
                )

                self.results.append(result)

            except Exception as e:
                self.results.append(QualityResult(
                    rule_name=rule.name,
                    passed=False,
                    total_records=total_records,
                    failed_records=-1,
                    pass_rate=0.0,
                    severity=rule.severity,
                    timestamp=datetime.now(),
                    details={"error": str(e)}
                ))

        return self.get_summary()

    def get_summary(self) -> Dict[str, Any]:
        """Get validation summary."""

        total_rules = len(self.results)
        passed_rules = sum(1 for r in self.results if r.passed)

        critical_failures = [
            r for r in self.results
            if not r.passed and r.severity == "critical"
        ]

        return {
            "timestamp": datetime.now().isoformat(),
            "total_rules": total_rules,
            "passed_rules": passed_rules,
            "failed_rules": total_rules - passed_rules,
            "pass_rate": passed_rules / total_rules * 100 if total_rules > 0 else 0,
            "critical_failures": len(critical_failures),
            "has_critical_failures": len(critical_failures) > 0,
            "results": [
                {
                    "rule": r.rule_name,
                    "passed": r.passed,
                    "pass_rate": r.pass_rate,
                    "failed_records": r.failed_records,
                    "severity": r.severity
                }
                for r in self.results
            ]
        }

    def save_results(self, output_path: str) -> None:
        """Save validation results to Delta table."""

        results_data = [
            {
                "rule_name": r.rule_name,
                "passed": r.passed,
                "total_records": r.total_records,
                "failed_records": r.failed_records,
                "pass_rate": r.pass_rate,
                "severity": r.severity,
                "timestamp": r.timestamp,
                "details": json.dumps(r.details)
            }
            for r in self.results
        ]

        results_df = self.spark.createDataFrame(results_data)
        results_df.write.format("delta").mode("append").save(output_path)

"""
DLT pipeline with comprehensive quality monitoring.
"""
import dlt
from pyspark.sql.functions import *
from typing import Dict, Any


class DLTQualityMonitor:
    """Monitor data quality in DLT pipelines."""

    @staticmethod
    def create_quality_tables():
        """Create tables with quality monitoring."""

        @dlt.table(
            name="quality_metrics",
            comment="Quality metrics for all pipeline tables",
            table_properties={
                "quality": "monitoring",
                "delta.enableChangeDataFeed": "true"
            }
        )
        def quality_metrics():
            """Aggregate quality metrics across pipeline."""
            return spark.sql("""
                SELECT
                    current_timestamp() as metric_timestamp,
                    'bronze_orders' as table_name,
                    COUNT(*) as total_records,
                    COUNT_IF(order_id IS NULL) as null_order_ids,
                    COUNT_IF(amount <= 0) as invalid_amounts,
                    COUNT_IF(order_date > current_date()) as future_dates
                FROM LIVE.bronze_orders
            """)

        @dlt.table(
            name="orders_with_quality",
            comment="Orders with quality scores"
        )
        @dlt.expect("valid_order_id", "order_id IS NOT NULL")
        @dlt.expect_or_drop("positive_amount", "amount > 0")
        @dlt.expect("valid_email", "email RLIKE '^[A-Za-z0-9._%+-]+@'")
        def orders_with_quality():
            return (
                dlt.read_stream("bronze_orders")
                .withColumn(
                    "quality_score",
                    when(col("order_id").isNotNull(), 1).otherwise(0) +
                    when(col("amount") > 0, 1).otherwise(0) +
                    when(col("email").rlike("^[A-Za-z0-9._%+-]+@"), 1).otherwise(0)
                )
                .withColumn(
                    "quality_tier",
                    when(col("quality_score") >= 3, "high")
                    .when(col("quality_score") >= 2, "medium")
                    .otherwise("low")
                )
            )

    @staticmethod
    def get_quality_report(event_log_path: str) -> Dict[str, Any]:
        """Generate quality report from DLT event log."""

        events = spark.read.format("delta").load(event_log_path)

        # Extract expectation metrics
        expectations = (
            events
            .filter(col("event_type") == "flow_progress")
            .select(
                col("timestamp"),
                col("origin.flow_name").alias("table_name"),
                col("details.flow_progress.metrics.num_output_rows").alias("output_rows"),
                explode(col("details.flow_progress.data_quality.expectations")).alias("expectation")
            )
            .select(
                "timestamp",
                "table_name",
                "output_rows",
                col("expectation.name").alias("expectation_name"),
                col("expectation.passed_records").alias("passed"),
                col("expectation.failed_records").alias("failed")
            )
        )

        return {
            "total_tables": expectations.select("table_name").distinct().count(),
            "total_expectations": expectations.count(),
            "failed_expectations": expectations.filter(col("failed") > 0).count(),
            "details": expectations.collect()
        }