Best Practices for Data Lake Governance¶

# Schema validation at ingestion
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, TimestampType
from pyspark.sql.functions import current_timestamp, col, when

def enforce_data_quality(spark, raw_data_path, governed_path):
    """Enforce data quality at ingestion"""

    # Define strict schema
    schema = StructType([
        StructField("id", StringType(), False),
        StructField("timestamp", TimestampType(), False),
        StructField("location", StringType(), True),
        StructField("value", StringType(), True)
    ])

    # Read with schema validation
    df = spark.read.schema(schema).parquet(raw_data_path)

    # Apply quality checks
    df_quality = df.withColumn(
        "quality_score",
        when(col("id").isNull(), 0.0)
        .when(col("timestamp").isNull(), 0.0)
        .otherwise(1.0)
    ).withColumn(
        "ingest_time", current_timestamp()
    )

    # Write to governed zone
    df_quality.write \
        .format("delta") \
        .option("mergeSchema", "false") \
        .option("delta.autoOptimize.optimizeWrite", "true") \
        .save(governed_path)

    return df_quality

# Partitioning strategy for governance
def create_governed_table(spark, table_name, schema):
    """Create governed table with proper partitioning"""

    spark.sql(f"""
        CREATE TABLE {table_name} (
            {schema}
        )
        USING DELTA
        PARTITIONED BY (year, month, day)
        CLUSTERED BY (id) INTO 100 BUCKETS
        TBLPROPERTIES (
            'delta.autoOptimize.optimizeWrite' = 'true',
            'delta.autoOptimize.autoCompact' = 'true',
            'delta.enableChangeDataFeed' = 'true'
        )
    """)

    # Register in Glue catalog
    spark.sql(f"""
        CREATE EXTERNAL TABLE {table_name}_external
        STORED AS PARQUET
        LOCATION 's3://data-lake/governed/{table_name}/'
        TBLPROPERTIES (
            'classification' = 'governed',
            'owner' = 'data-engineering',
            'compliance' = 'gdpr-ready'
        )
    """)

# Delta Lake versioning for compliance
def create_governed_delta_table(spark, table_name):
    """Create Delta table with versioning and time travel"""

    # Create table with versioning
    spark.sql(f"""
        CREATE TABLE {table_name} (
            id STRING,
            data STRING,
            ingest_time TIMESTAMP,
            version INT
        )
        USING DELTA
        TBLPROPERTIES (
            'delta.logRetentionDuration' = 'interval 7 days',
            'delta.deletedFileRetentionDuration' = 'interval 7 days',
            'delta.enableChangeDataFeed' = 'true'
        )
    """)

    # Enable time travel for compliance
    spark.sql(f"""
        ALTER TABLE {table_name} SET TBLPROPERTIES (
            'delta.retentionDurationCheck.enabled' = 'false'
        )
    """)

# Metadata catalog implementation
class DataLakeCatalog:
    def __init__(self, spark, catalog_name="data_lake"):
        self.spark = spark
        self.catalog_name = catalog_name

    def register_object(self, object_path, metadata):
        """Register object in metadata catalog"""

        # Extract object metadata
        object_id = self._generate_object_id(object_path)
        schema_version = metadata.get("schema_version", "1.0")
        owner = metadata.get("owner", "unknown")
        classification = metadata.get("classification", "INTERNAL")

        # Insert into catalog
        self.spark.sql(f"""
            INSERT INTO {self.catalog_name}.metadata_catalog
            VALUES (
                '{object_id}',
                '{object_path}',
                '{metadata.get("object_type", "table")}',
                '{schema_version}',
                '{owner}',
                current_timestamp(),
                current_timestamp(),
                '{classification}',
                '{json.dumps(metadata.get("compliance_flags", {}))}',
                '{json.dumps(metadata.get("lineage_parents", []))}',
                '{json.dumps(metadata.get("lineage_children", []))}'
            )
        """)

    def track_lineage(self, source_path, target_path, transformation):
        """Track data lineage"""

        lineage_id = self._generate_lineage_id(source_path, target_path)

        self.spark.sql(f"""
            INSERT INTO {self.catalog_name}.data_lineage
            VALUES (
                '{lineage_id}',
                '{source_path}',
                '{target_path}',
                '{transformation}',
                current_timestamp(),
                '{{}}'
            )
        """)

    def _generate_object_id(self, path):
        """Generate unique object ID"""
        import hashlib
        return hashlib.md5(path.encode()).hexdigest()

    def _generate_lineage_id(self, source, target):
        """Generate unique lineage ID"""
        import hashlib
        return hashlib.md5(f"{source}->{target}".encode()).hexdigest()

# S3 Object Lock configuration
def configure_object_lock(bucket_name, retention_period=2555):  # 7 years
    """Configure S3 Object Lock for legal compliance"""

    import boto3

    s3_client = boto3.client('s3')

    # Enable Object Lock
    s3_client.put_object_lock_configuration(
        Bucket=bucket_name,
        ObjectLockConfiguration={
            'ObjectLockEnabled': 'Enabled',
            'Rule': {
                'DefaultRetention': {
                    'Mode': 'GOVERNANCE',
                    'Days': retention_period
                }
            }
        }
    )

    # Apply to specific objects
    s3_client.put_object(
        Bucket=bucket_name,
        Key='governed/legal-hold/data.parquet',
        Body=data,
        ObjectLockMode='GOVERNANCE',
        ObjectLockRetainUntilDate=datetime(2030, 12, 31),
        ObjectLockLegalHoldStatus='ON'
    )

# PII/PHI tagging at ingestion
def tag_sensitive_data(spark, df, classification_rules):
    """Tag sensitive data at ingestion"""

    from pyspark.sql.functions import when, col, lit

    # Apply classification rules
    df_tagged = df
    for rule in classification_rules:
        if rule["type"] == "pii":
            df_tagged = df_tagged.withColumn(
                "pii_flag",
                when(col(rule["column"]).rlike(rule["pattern"]), lit(True))
                .otherwise(lit(False))
            )
        elif rule["type"] == "phi":
            df_tagged = df_tagged.withColumn(
                "phi_flag",
                when(col(rule["column"]).rlike(rule["pattern"]), lit(True))
                .otherwise(lit(False))
            )

    # Add compliance metadata
    df_tagged = df_tagged.withColumn(
        "compliance_flags",
        lit({
            "gdpr_applicable": True,
            "hipaa_applicable": True,
            "retention_period": 2555,  # 7 years
            "legal_hold": False
        })
    )

    return df_tagged

# Immutable ingestion pattern
def immutable_ingestion(spark, raw_data_path, governed_path):
    """Implement immutable ingestion pattern"""

    # Read raw data
    df = spark.read.parquet(raw_data_path)

    # Add governance metadata
    df_governed = df.withColumn(
        "ingest_time", current_timestamp()
    ).withColumn(
        "ingest_id", lit(uuid.uuid4().hex)
    ).withColumn(
        "schema_version", lit("1.0")
    )

    # Write to immutable zone
    df_governed.write \
        .format("delta") \
        .option("mergeSchema", "false") \
        .option("delta.autoOptimize.optimizeWrite", "true") \
        .mode("append") \
        .save(governed_path)

    # Register in catalog
    catalog.register_object(
        object_path=governed_path,
        metadata={
            "object_type": "table",
            "owner": "data-engineering",
            "classification": "INTERNAL",
            "compliance_flags": {
                "immutable": True,
                "retention_period": 2555
            }
        }
    )

# Automated quality validation
def validate_data_quality(spark, table_name, quality_rules):
    """Validate data quality against rules"""

    validation_results = {}

    for rule in quality_rules:
        if rule["type"] == "row_count":
            count = spark.table(table_name).count()
            validation_results[rule["name"]] = {
                "actual": count,
                "expected": rule["expected"],
                "passed": count >= rule["expected"]
            }

        elif rule["type"] == "null_threshold":
            null_count = spark.table(table_name).filter(
                col(rule["column"]).isNull()
            ).count()
            total_count = spark.table(table_name).count()
            null_percentage = null_count / total_count

            validation_results[rule["name"]] = {
                "actual": null_percentage,
                "threshold": rule["threshold"],
                "passed": null_percentage <= rule["threshold"]
            }

        elif rule["type"] == "geospatial_validity":
            invalid_geom_count = spark.table(table_name).filter(
                ~col(rule["column"]).rlike("^POINT\\(-?\\d+\\.\\d+ -?\\d+\\.\\d+\\)$")
            ).count()

            validation_results[rule["name"]] = {
                "invalid_geometries": invalid_geom_count,
                "passed": invalid_geom_count == 0
            }

    return validation_results

# Stress test concurrent queries
def stress_test_concurrent_queries(spark, table_name, num_queries=100):
    """Stress test concurrent BI queries"""

    import concurrent.futures
    import time

    def execute_query(query_id):
        """Execute single query"""
        start_time = time.time()

        # Simulate BI query
        result = spark.sql(f"""
            SELECT 
                partition_col,
                COUNT(*) as record_count,
                AVG(numeric_col) as avg_value
            FROM {table_name}
            WHERE partition_col >= '2024-01-01'
            GROUP BY partition_col
            ORDER BY partition_col
        """).collect()

        runtime = time.time() - start_time
        return {
            "query_id": query_id,
            "runtime": runtime,
            "result_count": len(result)
        }

    # Execute concurrent queries
    with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
        futures = [executor.submit(execute_query, i) for i in range(num_queries)]
        results = [future.result() for future in concurrent.futures.as_completed(futures)]

    # Analyze results
    avg_runtime = sum(r["runtime"] for r in results) / len(results)
    max_runtime = max(r["runtime"] for r in results)

    print(f"Concurrent queries: {num_queries}")
    print(f"Average runtime: {avg_runtime:.2f}s")
    print(f"Max runtime: {max_runtime:.2f}s")

    return results

# Multi-region governance reconciliation
def reconcile_multi_region_governance(spark, regions):
    """Reconcile governance across regions"""

    reconciliation_results = {}

    for region in regions:
        # Get governance metadata from each region
        region_metadata = spark.sql(f"""
            SELECT 
                object_id,
                object_path,
                classification,
                compliance_flags,
                updated_at
            FROM {region}.metadata_catalog
            WHERE updated_at >= current_timestamp() - interval 1 hour
        """).collect()

        reconciliation_results[region] = region_metadata

    # Detect conflicts
    conflicts = []
    for obj_id in set(obj["object_id"] for region_data in reconciliation_results.values() for obj in region_data):
        region_versions = {}
        for region, metadata in reconciliation_results.items():
            for obj in metadata:
                if obj["object_id"] == obj_id:
                    region_versions[region] = obj

        # Check for conflicts
        if len(set(obj["classification"] for obj in region_versions.values())) > 1:
            conflicts.append({
                "object_id": obj_id,
                "conflicting_regions": list(region_versions.keys()),
                "classifications": [obj["classification"] for obj in region_versions.values()]
            })

    return conflicts

# GDPR right to be forgotten
def gdpr_data_deletion(spark, subject_id, deletion_request):
    """Implement GDPR data deletion"""

    # Find all data related to subject
    subject_data = spark.sql(f"""
        SELECT 
            object_id,
            object_path,
            classification,
            compliance_flags
        FROM metadata_catalog
        WHERE compliance_flags.pii_subject_id = '{subject_id}'
        OR object_path LIKE '%{subject_id}%'
    """).collect()

    deletion_results = []

    for data_obj in subject_data:
        # Check if deletion is allowed
        if data_obj["classification"] == "LEGAL_HOLD":
            deletion_results.append({
                "object_id": data_obj["object_id"],
                "status": "BLOCKED",
                "reason": "Legal hold prevents deletion"
            })
            continue

        # Perform deletion
        try:
            # Delete from storage
            spark.sql(f"DROP TABLE IF EXISTS {data_obj['object_path']}")

            # Update metadata
            spark.sql(f"""
                UPDATE metadata_catalog
                SET 
                    updated_at = current_timestamp(),
                    compliance_flags = jsonb_set(
                        compliance_flags,
                        '{{deleted}}',
                        'true'
                    )
                WHERE object_id = '{data_obj['object_id']}'
            """)

            deletion_results.append({
                "object_id": data_obj["object_id"],
                "status": "DELETED",
                "timestamp": datetime.now().isoformat()
            })

        except Exception as e:
            deletion_results.append({
                "object_id": data_obj["object_id"],
                "status": "ERROR",
                "error": str(e)
            })

    return deletion_results

# ❌ WRONG: No governance at ingestion
def bad_ingestion_pattern():
    """Don't ingest without governance"""
    # Raw data goes directly to data lake
    df = spark.read.parquet("s3://raw-data/")
    df.write.parquet("s3://data-lake/")  # No governance!

# ✅ CORRECT: Governance-first ingestion
def good_ingestion_pattern():
    """Governance-first ingestion"""
    # Apply governance at ingestion
    df = spark.read.parquet("s3://raw-data/")
    df_governed = apply_governance(df)
    df_governed.write.parquet("s3://data-lake/governed/")
    register_in_catalog(df_governed)

# ❌ WRONG: No lineage tracking
def bad_lineage_pattern():
    """Don't ignore lineage"""
    # Transform data without tracking lineage
    df1 = spark.read.table("source")
    df2 = df1.filter(col("status") == "active")
    df2.write.saveAsTable("target")  # No lineage!

# ✅ CORRECT: Lineage tracking
def good_lineage_pattern():
    """Track lineage for all transformations"""
    df1 = spark.read.table("source")
    df2 = df1.filter(col("status") == "active")
    df2.write.saveAsTable("target")
    track_lineage("source", "target", "filter_active")

# Governance maturity assessment
def assess_governance_maturity():
    """Assess governance maturity level"""

    maturity_levels = {
        "level_1": {
            "name": "Basic",
            "description": "Manual governance processes",
            "indicators": [
                "Manual schema validation",
                "Basic access controls",
                "Limited metadata"
            ]
        },
        "level_2": {
            "name": "Managed",
            "description": "Automated governance processes",
            "indicators": [
                "Automated schema validation",
                "Role-based access controls",
                "Centralized metadata catalog"
            ]
        },
        "level_3": {
            "name": "Optimized",
            "description": "Intelligent governance processes",
            "indicators": [
                "AI-powered data classification",
                "Dynamic access controls",
                "Automated compliance reporting"
            ]
        }
    }

    return maturity_levels