Data Quality SLAs, Validation Layers, and Observability for Tabular, Geospatial, and ML Data: Best Practices¶

# Schema validation
from pydantic import BaseModel, validator

class UserSchema(BaseModel):
    id: int
    email: str
    age: int

    @validator('email')
    def validate_email(cls, v):
        if '@' not in v:
            raise ValueError('Invalid email')
        return v

    @validator('age')
    def validate_age(cls, v):
        if v < 0 or v > 150:
            raise ValueError('Invalid age')
        return v

# Statistical validation
import pandas as pd
from scipy import stats

class StatisticalValidator:
    def validate(self, data: pd.DataFrame) -> ValidationResult:
        """Validate statistical properties"""
        results = []

        # Check for outliers
        for column in data.columns:
            z_scores = stats.zscore(data[column])
            outliers = (abs(z_scores) > 3).sum()

            if outliers > len(data) * 0.05:  # >5% outliers
                results.append(ValidationIssue(
                    column=column,
                    issue="excessive_outliers",
                    severity="warning"
                ))

        return ValidationResult(issues=results)

# Semantic validation
class SemanticValidator:
    def validate(self, data: pd.DataFrame) -> ValidationResult:
        """Validate semantic properties"""
        results = []

        # Check business rules
        if 'order_date' in data.columns and 'ship_date' in data.columns:
            invalid = data[data['ship_date'] < data['order_date']]
            if len(invalid) > 0:
                results.append(ValidationIssue(
                    issue="ship_date_before_order_date",
                    severity="error",
                    count=len(invalid)
                ))

        return ValidationResult(issues=results)

# Great Expectations validation
import great_expectations as ge

context = ge.get_context()

# Define expectations
expectation_suite = context.create_expectation_suite("my_suite")

# Add expectations
validator = context.get_validator(
    batch_request={
        "datasource_name": "my_datasource",
        "data_connector_name": "default_inferred_data_connector_name",
        "data_asset_name": "my_table"
    },
    expectation_suite_name="my_suite"
)

validator.expect_column_values_to_not_be_null("id")
validator.expect_column_values_to_be_unique("id")
validator.expect_column_values_to_be_between("age", min_value=0, max_value=150)

# Tile quality validation
class TileQualityValidator:
    def validate_tile(self, tile: Tile) -> QualityReport:
        """Validate tile quality"""
        issues = []

        # Check tile completeness
        if tile.coverage < 0.95:
            issues.append(QualityIssue(
                type="incomplete_coverage",
                severity="warning"
            ))

        # Check tile format
        if not tile.is_valid_format():
            issues.append(QualityIssue(
                type="invalid_format",
                severity="error"
            ))

        # Check tile bounds
        if not tile.is_valid_bounds():
            issues.append(QualityIssue(
                type="invalid_bounds",
                severity="error"
            ))

        return QualityReport(issues=issues)

# Raster quality validation
import rasterio

class RasterQualityValidator:
    def validate_raster(self, raster_path: str) -> QualityReport:
        """Validate raster quality"""
        issues = []

        with rasterio.open(raster_path) as src:
            # Check CRS
            if src.crs is None:
                issues.append(QualityIssue(
                    type="missing_crs",
                    severity="error"
                ))

            # Check data type
            if src.dtypes[0] not in ['uint8', 'uint16', 'float32']:
                issues.append(QualityIssue(
                    type="invalid_data_type",
                    severity="warning"
                ))

            # Check no-data values
            if src.nodata is None:
                issues.append(QualityIssue(
                    type="missing_nodata",
                    severity="warning"
                ))

        return QualityReport(issues=issues)

# ML feature drift detection
from evidently import ColumnMapping
from evidently.metric_preset import DataDriftPreset
from evidently.report import Report

class FeatureDriftDetector:
    def detect_drift(self, reference: pd.DataFrame, current: pd.DataFrame) -> DriftReport:
        """Detect feature drift"""
        # Define column mapping
        column_mapping = ColumnMapping(
            target=None,
            numerical_features=['age', 'income'],
            categorical_features=['category']
        )

        # Generate drift report
        report = Report(metrics=[DataDriftPreset()])
        report.run(
            reference_data=reference,
            current_data=current,
            column_mapping=column_mapping
        )

        # Extract drift metrics
        drift_metrics = report.as_dict()['metrics']

        return DriftReport(metrics=drift_metrics)

# DQ stability fitness function
class DQStabilityFitnessFunction:
    def evaluate(self, system: System) -> float:
        """Evaluate DQ stability"""
        # Calculate quality variance
        quality_scores = self.get_quality_scores(system)
        quality_variance = np.var(quality_scores)

        # Calculate stability (lower variance = higher stability)
        if quality_variance == 0:
            stability = 1.0
        else:
            stability = 1.0 / (1.0 + quality_variance)

        return stability

# Bad: No validation
def process_data(data):
    return process(data)  # No quality checks

# Good: Validation
def process_data(data):
    validator = DataQualityValidator()
    if not validator.validate(data):
        raise ValueError("Data quality violation")
    return process(data)