Interview Question

"Design a data quality framework that: (1) validates data at ingestion, (2) monitors data freshness, (3) detects anomalies, (4) quarantines bad data, (5) alerts stakeholders. How do you measure data quality, and what metrics do you track?"

Difficulty: Hard | Frequently asked at Google, Airbnb, Uber, Netflix

Theoretical Foundation

Data Quality Dimensions

Data quality is measured across multiple dimensions:

┌─────────────────────────────────────────────────────────────┐
│              Data Quality Dimensions                        │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  1. Completeness: Are all expected records present?         │
│     Metric: % of non-null values                           │
│     Formula: (non-null count / total count) × 100          │
│                                                             │
│  2. Accuracy: Does data reflect reality?                    │
│     Metric: % of values matching source of truth           │
│     Formula: (correct count / total count) × 100           │
│                                                             │
│  3. Consistency: Is data consistent across systems?         │
│     Metric: % of values matching across sources            │
│     Formula: (matching values / total values) × 100        │
│                                                             │
│  4. Timeliness: Is data available when needed?              │
│     Metric: Time from event to availability                │
│     Formula: availability_time - event_time                │
│                                                             │
│  5. Uniqueness: Are records unique?                         │
│     Metric: % of unique records                            │
│     Formula: (unique count / total count) × 100            │
│                                                             │
│  6. Validity: Does data conform to rules?                   │
│     Metric: % of values passing validation rules           │
│     Formula: (valid count / total count) × 100             │
│                                                             │
└─────────────────────────────────────────────────────────────┘

Data Quality Score

Where:

• $C$ = Completeness score
• $A$ = Accuracy score
• $Co$ = Consistency score
• $T$ = Timeliness score
• $U$ = Uniqueness score
• $V$ = Validity score
• $w_i$ = Weights (sum to 1)

Example weights for different use cases:

Data Quality Pipeline

Validation Rules

Schema Validation

schema_rules = {
    "columns": {
        "user_id": {"type": "string", "nullable": False},
        "email": {"type": "string", "nullable": False, "pattern": r"^[\w\.-]+@[\w\.-]+\.\w+$"},
        "age": {"type": "integer", "nullable": True, "min": 0, "max": 150},
        "signup_date": {"type": "date", "nullable": False},
    }
}

Statistical Validation

statistical_rules = {
    "row_count": {
        "min": 1000,
        "max": 10000000,
        "expected": 50000,
        "tolerance": 0.2  # ±20%
    },
    "null_rate": {
        "user_id": {"max": 0.0},
        "email": {"max": 0.01},
        "age": {"max": 0.05},
    }
}

Business Rules

business_rules = [
    # Foreign key relationships
    {"type": "referential", "table": "orders", "column": "user_id", 
     "references": {"table": "users", "column": "id"}},
    
    # Cross-column validation
    {"type": "expression", "condition": "start_date <= end_date"},
    
    # Business logic
    {"type": "custom", "function": "validate_email_domain"},
]

Anomaly Detection Methods

1. Statistical Process Control (SPC)

Points outside control limits are anomalies.

2. Z-Score

If $|z| > 3$, the value is anomalous.

3. IQR Method

4. Time Series Decomposition

Where:

• $T_t$ = Trend component
• $S_t$ = Seasonal component
• $R_t$ = Residual (anomaly if large)

Data Freshness

┌─────────────────────────────────────────────────────────────┐
│              Data Freshness Metrics                         │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  1. Latency: Time from event to availability               │
│     Formula: available_time - event_time                    │
│     Target: < 1 hour (batch), < 1 minute (streaming)      │
│                                                             │
│  2. Freshness: Time since last update                      │
│     Formula: now - last_update_time                        │
│     Target: < schedule_interval + buffer                   │
│                                                             │
│  3. Staleness: How outdated the data is                    │
│     Formula: now - max(timestamp_column)                   │
│     Target: < business_requirement                         │
│                                                             │
└─────────────────────────────────────────────────────────────┘

Code Implementation

Data Quality Framework Core

from dataclasses import dataclass
from typing import List, Dict, Callable, Optional
from datetime import datetime, timedelta
import pandas as pd
import numpy as np
from enum import Enum

class QualityCheckResult(Enum):
    PASS = "pass"
    FAIL = "fail"
    WARN = "warn"

@dataclass
class QualityCheck:
    name: str
    description: str
    check_func: Callable
    severity: str  # critical, warning, info
    dimensions: List[str]  # completeness, accuracy, etc.

@dataclass
class QualityReport:
    timestamp: datetime
    table_name: str
    checks: List[Dict]
    overall_score: float
    passed: bool
    details: Dict

class DataQualityFramework:
    def __init__(self, spark_session=None):
        self.spark = spark_session
        self.checks = {}
        self.reports = []
        
    def register_check(self, check: QualityCheck):
        """Register a quality check"""
        self.checks[check.name] = check
    
    def run_checks(self, df, table_name: str) -> QualityReport:
        """Run all registered checks on a DataFrame"""
        
        results = []
        scores = {}
        
        for name, check in self.checks.items():
            try:
                result = check.check_func(df)
                results.append({
                    'name': name,
                    'description': check.description,
                    'result': result,
                    'severity': check.severity,
                    'dimensions': check.dimensions,
                })
                
                # Calculate dimension scores
                for dim in check.dimensions:
                    if dim not in scores:
                        scores[dim] = []
                    scores[dim].append(1.0 if result == QualityCheckResult.PASS else 0.0)
                    
            except Exception as e:
                results.append({
                    'name': name,
                    'description': check.description,
                    'result': QualityCheckResult.FAIL,
                    'severity': check.severity,
                    'error': str(e),
                })
        
        # Calculate overall score
        all_scores = []
        for dim, dim_scores in scores.items():
            all_scores.append(np.mean(dim_scores))
        
        overall_score = np.mean(all_scores) if all_scores else 0.0
        
        # Determine if passed (no critical failures)
        critical_failures = [r for r in results 
                           if r['result'] == QualityCheckResult.FAIL 
                           and r['severity'] == 'critical']
        
        report = QualityReport(
            timestamp=datetime.now(),
            table_name=table_name,
            checks=results,
            overall_score=overall_score,
            passed=len(critical_failures) == 0,
            details=scores,
        )
        
        self.reports.append(report)
        return report

Built-in Quality Checks

class BuiltinChecks:
    """Built-in quality checks"""
    
    @staticmethod
    def completeness_check(columns: List[str], max_null_rate: float = 0.0):
        """Check completeness of specified columns"""
        
        def check(df):
            for col in columns:
                null_rate = df.filter(df[col].isNull()).count() / df.count()
                if null_rate > max_null_rate:
                    return QualityCheckResult.FAIL
            return QualityCheckResult.PASS
        
        return QualityCheck(
            name=f"completeness_{'_'.join(columns)}",
            description=f"Check completeness of {columns}",
            check_func=check,
            severity='critical',
            dimensions=['completeness'],
        )
    
    @staticmethod
    def uniqueness_check(columns: List[str]):
        """Check uniqueness of specified columns"""
        
        def check(df):
            total_count = df.count()
            distinct_count = df.select(columns).distinct().count()
            
            if total_count != distinct_count:
                return QualityCheckResult.FAIL
            return QualityCheckResult.PASS
        
        return QualityCheck(
            name=f"uniqueness_{'_'.join(columns)}",
            description=f"Check uniqueness of {columns}",
            check_func=check,
            severity='critical',
            dimensions=['uniqueness'],
        )
    
    @staticmethod
    def range_check(column: str, min_val=None, max_val=None):
        """Check if values are within range"""
        
        def check(df):
            if min_val is not None:
                below_min = df.filter(df[col] < min_val).count()
                if below_min > 0:
                    return QualityCheckResult.FAIL
            
            if max_val is not None:
                above_max = df.filter(df[col] > max_val).count()
                if above_max > 0:
                    return QualityCheckResult.FAIL
            
            return QualityCheckResult.PASS
        
        return QualityCheck(
            name=f"range_{column}",
            description=f"Check range of {column}",
            check_func=check,
            severity='warning',
            dimensions=['validity'],
        )
    
    @staticmethod
    def pattern_check(column: str, pattern: str):
        """Check if values match pattern"""
        
        def check(df):
            import re
            non_matching = df.filter(~df[col].rlike(pattern)).count()
            if non_matching > 0:
                return QualityCheckResult.FAIL
            return QualityCheckResult.PASS
        
        return QualityCheck(
            name=f"pattern_{column}",
            description=f"Check pattern of {column}",
            check_func=check,
            severity='warning',
            dimensions=['validity'],
        )
    
    @staticmethod
    def freshness_check(timestamp_column: str, max_age_hours: int = 24):
        """Check data freshness"""
        
        def check(df):
            from pyspark.sql import functions as F
            max_timestamp = df.agg(F.max(timestamp_column)).collect()[0][0]
            
            if max_timestamp is None:
                return QualityCheckResult.FAIL
            
            age = datetime.now() - max_timestamp
            if age > timedelta(hours=max_age_hours):
                return QualityCheckResult.FAIL
            
            return QualityCheckResult.PASS
        
        return QualityCheck(
            name=f"freshness_{timestamp_column}",
            description=f"Check freshness of {timestamp_column}",
            check_func=check,
            severity='critical',
            dimensions=['timeliness'],
        )
    
    @staticmethod
    def row_count_check(min_rows: int, max_rows: int = None):
        """Check row count is within bounds"""
        
        def check(df):
            count = df.count()
            if count < min_rows:
                return QualityCheckResult.FAIL
            if max_rows and count > max_rows:
                return QualityCheckResult.FAIL
            return QualityCheckResult.PASS
        
        return QualityCheck(
            name="row_count",
            description=f"Check row count between {min_rows} and {max_rows}",
            check_func=check,
            severity='critical',
            dimensions=['completeness'],
        )

Anomaly Detection

class AnomalyDetector:
    """Detect anomalies in data"""
    
    def __init__(self, spark_session):
        self.spark = spark_session
    
    def detect_statistical_anomalies(self, df, column, threshold=3.0):
        """Detect anomalies using Z-score"""
        from pyspark.sql import functions as F
        
        # Calculate mean and std
        stats = df.agg(
            F.mean(column).alias('mean'),
            F.stddev(column).alias('stddev')
        ).collect()[0]
        
        mean = stats['mean']
        stddev = stats['stddev']
        
        # Calculate Z-score
        anomalies = df.withColumn(
            'z_score',
            (F.col(column) - mean) / stddev
        ).filter(F.abs(F.col('z_score')) > threshold)
        
        return anomalies
    
    def detect_iqr_anomalies(self, df, column):
        """Detect anomalies using IQR method"""
        from pyspark.sql import functions as F
        
        # Calculate quartiles
        quantiles = df.approxQuantile(column, [0.25, 0.75], 0.01)
        q1, q3 = quantiles
        
        iqr = q3 - q1
        lower_bound = q1 - 1.5 * iqr
        upper_bound = q3 + 1.5 * iqr
        
        # Find anomalies
        anomalies = df.filter(
            (F.col(column) < lower_bound) | 
            (F.col(column) > upper_bound)
        )
        
        return anomalies
    
    def detect_time_series_anomalies(self, df, timestamp_col, value_col, 
                                      window_size=7, threshold=2.0):
        """Detect anomalies in time series data"""
        from pyspark.sql import functions as F
        from pyspark.sql.window import Window
        
        # Calculate rolling average and std
        window = Window.orderBy(timestamp_col).rowsBetween(-window_size, -1)
        
        df_with_stats = df.withColumn(
            'rolling_avg', F.avg(value_col).over(window)
        ).withColumn(
            'rolling_std', F.stddev(value_col).over(window)
        )
        
        # Detect anomalies
        anomalies = df_with_stats.withColumn(
            'z_score',
            (F.col(value_col) - F.col('rolling_avg')) / F.col('rolling_std')
        ).filter(F.abs(F.col('z_score')) > threshold)
        
        return anomalies

Data Quarantine

class DataQuarantine:
    """Quarantine bad data for investigation"""
    
    def __init__(self, spark_session, quarantine_path):
        self.spark = spark_session
        self.quarantine_path = quarantine_path
    
    def quarantine_records(self, bad_records, reason, table_name):
        """Quarantine bad records"""
        from pyspark.sql import functions as F
        
        # Add metadata
        quarantined = bad_records \
            .withColumn('_quarantine_reason', F.lit(reason)) \
            .withColumn('_quarantine_timestamp', F.current_timestamp()) \
            .withColumn('_source_table', F.lit(table_name))
        
        # Write to quarantine storage
        quarantined.write \
            .format("delta") \
            .mode("append") \
            .partitionBy("_quarantine_timestamp") \
            .save(f"{self.quarantine_path}/{table_name}")
        
        return quarantined.count()
    
    def get_quarantine_stats(self, table_name):
        """Get quarantine statistics"""
        from pyspark.sql import functions as F
        
        quarantine_df = self.spark.read \
            .format("delta") \
            .load(f"{self.quarantine_path}/{table_name}")
        
        stats = quarantine_df.groupBy('_quarantine_reason') \
            .agg(
                F.count("*").alias("count"),
                F.min("_quarantine_timestamp").alias("first_seen"),
                F.max("_quarantine_timestamp").alias("last_seen")
            )
        
        return stats

Monitoring and Alerting

class DataQualityMonitor:
    """Monitor data quality and send alerts"""
    
    def __init__(self, alert_config):
        self.alert_config = alert_config
    
    def send_alert(self, report: QualityReport):
        """Send alert based on quality report"""
        
        if not report.passed:
            self._send_critical_alert(report)
        elif report.overall_score < 0.9:
            self._send_warning_alert(report)
    
    def _send_critical_alert(self, report):
        """Send critical alert"""
        import requests
        
        critical_failures = [c for c in report.checks 
                           if c['result'] == QualityCheckResult.FAIL 
                           and c['severity'] == 'critical']
        
        message = f"""
🚨 CRITICAL DATA QUALITY FAILURE

Table: {report.table_name}
Time: {report.timestamp}
Score: {report.overall_score:.2%}

Failed Checks:
{chr(10).join(f"- {c['name']}: {c.get('error', 'Failed')}" for c in critical_failures)}

Please investigate immediately.
        """
        
        # Send to Slack/PagerDuty
        requests.post(
            self.alert_config['slack_webhook'],
            json={'text': message}
        )
    
    def _send_warning_alert(self, report):
        """Send warning alert"""
        import requests
        
        message = f"""
⚠️ Data Quality Warning

Table: {report.table_name}
Time: {report.timestamp}
Score: {report.overall_score:.2%}

Some checks are below threshold.
        """
        
        requests.post(
            self.alert_config['slack_webhook'],
            json={'text': message}
        )

Complete Example

# ============================================================
# COMPLETE DATA QUALITY FRAMEWORK EXAMPLE
# ============================================================

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("DataQualityFramework") \
    .getOrCreate()

# Initialize framework
dq_framework = DataQualityFramework(spark)

# Register checks
dq_framework.register_check(
    BuiltinChecks.completeness_check(['user_id', 'email'], max_null_rate=0.01)
)

dq_framework.register_check(
    BuiltinChecks.uniqueness_check(['user_id'])
)

dq_framework.register_check(
    BuiltinChecks.range_check('age', min_val=0, max_val=150)
)

dq_framework.register_check(
    BuiltinChecks.freshness_check('updated_at', max_age_hours=24)
)

dq_framework.register_check(
    BuiltinChecks.row_count_check(min_rows=1000)
)

# Read data
df = spark.read.parquet("s3://data-lake/users/")

# Run quality checks
report = dq_framework.run_checks(df, "users")

# Process report
if not report.passed:
    # Quarantine bad data
    quarantine = DataQuarantine(spark, "s3://quarantine/")
    
    # Find failing records (example)
    bad_records = df.filter(df.user_id.isNull())
    quarantine.quarantine_records(bad_records, "null_user_id", "users")

# Send alerts
monitor = DataQualityMonitor({
    'slack_webhook': Variable.get('slack_webhook_url')
})
monitor.send_alert(report)

# Generate quality report
print(f"Overall Score: {report.overall_score:.2%}")
print(f"Passed: {report.passed}")
for check in report.checks:
    print(f"  {check['name']}: {check['result'].value}")

Great Expectations Integration

import great_expectations as ge

# ============================================================
# GREAT EXPECTATIONS INTEGRATION
# ============================================================

# Convert to Great Expectations DataFrame
ge_df = ge.from_pandas(df.toPandas())

# Define expectations
ge_df.expect_column_values_to_not_be_null('user_id')
ge_df.expect_column_values_to_be_unique('user_id')
ge_df.expect_column_values_to_be_between('age', min_value=0, max_value=150)
ge_df.expect_column_values_to_match_regex('email', r'^[\w\.-]+@[\w\.-]+\.\w+$')

# Validate
results = ge_df.validate()

# Check results
if not results.success:
    print("Validation failed!")
    for result in results.results:
        if not result.success:
            print(f"  {result.expectation_config.expectation_type}: {result.result}")

Common Follow-Up Questions

Q1: How do you measure data quality ROI?

Track metrics:

• Incidents prevented: Number of bad data incidents caught before impact
• Time saved: Hours spent on data debugging reduced
• Revenue protected: Business impact of data quality improvements
• Customer satisfaction: Reduction in data-related complaints

Q2: How do you handle data quality in real-time?

# Real-time quality checks with Spark Streaming
streaming_df = spark.readStream \
    .format("kafka") \
    .load()

# Apply quality checks
quality_checked = streaming_df \
    .withColumn("is_valid", 
        (F.col("user_id").isNotNull()) &
        (F.col("amount") > 0)
    )

# Split into good and bad streams
good_stream = quality_checked.filter(F.col("is_valid"))
bad_stream = quality_checked.filter(~F.col("is_valid"))

# Write to separate sinks
good_stream.writeStream.format("delta").start("s3://good-data/")
bad_stream.writeStream.format("delta").start("s3://quarantine/")

Q3: How do you handle data quality across teams?

• Data contracts: Define quality expectations between teams
• SLAs: Agree on freshness and quality targets
• Shared metrics: Common quality dashboards
• Escalation process: Clear ownership for quality issues

Q4: How do you prioritize quality checks?

Prioritize based on:

1. Business impact: What's the cost of bad data?
2. Frequency: How often does this issue occur?
3. Detection difficulty: How easy is it to detect?
4. Remediation cost: How hard is it to fix?

Company-Specific Tips

Google Interview Tips

• Discuss data contracts and schema evolution
• Explain anomaly detection at scale
• Mention data freshness monitoring
• Talk about automated remediation

Airbnb Interview Tips

• Focus on data trust and user-facing metrics
• Discuss A/B testing data quality
• Mention ML data quality for recommendations
• Talk about cost optimization for quality checks

Uber Interview Tips

• Discuss real-time data quality for ride matching
• Explain geospatial data validation
• Mention payment data accuracy
• Talk about multi-region quality monitoring