Data Lake Architecture: Designing Scalable Storage Systems

# Medallion Architecture Implementation with PySpark
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

spark = SparkSession.builder \
    .appName("MedallionArchitecture") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.SparkSessionExtensions") \
    .getOrCreate()

# ============================================================
# BRONZE LAYER: Raw Ingestion
# ============================================================
def ingest_to_bronze(spark, source_path, bronze_path):
    """
    Ingest raw data into Bronze layer.
    No transformations � preserve original format.
    """
    raw_df = spark.read \
        .format("json") \
        .option("multiLine", "true") \
        .load(source_path)

    raw_df.withColumn("_ingestion_timestamp", current_timestamp()) \
        .withColumn("_source_file", input_file_name()) \
        .withColumn("_load_date", current_date()) \
        .write \
        .format("parquet") \
        .mode("append") \
        .partitionBy("_load_date") \
        .save(bronze_path)

    print(f"Bronze: Ingested {raw_df.count()} records")

# ============================================================
# SILVER LAYER: Cleaned & Conformed
# ============================================================
def transform_to_silver(spark, bronze_path, silver_path):
    """
    Clean, deduplicate, validate, and enrich Bronze data.
    """
    bronze_df = spark.read.format("parquet").load(bronze_path)

    silver_df = bronze_df \
        .dropDuplicates(["order_id"]) \
        .filter(col("order_id").isNotNull()) \
        .withColumn("order_date", to_date(col("order_date"), "yyyy-MM-dd")) \
        .withColumn("amount", col("amount").cast(DecimalType(12, 2))) \
        .withColumn("customer_id", col("customer_id").cast(IntegerType())) \
        .withColumn("order_status",
            when(col("status").isin("complete", "completed"), "completed")
            .when(col("status").isin("pending", "processing"), "pending")
            .when(col("status").isin("cancel", "cancelled"), "cancelled")
            .otherwise("unknown")
        ) \
        .withColumn("_cleaned_timestamp", current_timestamp()) \
        .withColumn("_quality_score",
            when(col("email").isNotNull() & col("address").isNotNull(), 1.0)
            .when(col("email").isNotNull(), 0.7)
            .otherwise(0.3)
        )

    silver_df.write \
        .format("parquet") \
        .mode("overwrite") \
        .partitionBy("order_date") \
        .save(silver_path)

    print(f"Silver: Processed {silver_df.count()} records")

# ============================================================
# GOLD LAYER: Business-Ready Aggregations
# ============================================================
def aggregate_to_gold(spark, silver_path, gold_path):
    """
    Create business-ready aggregations and dimensions.
    """
    silver_df = spark.read.format("parquet").load(silver_path)

    # Daily revenue aggregation
    daily_metrics = silver_df \
        .groupBy("order_date", "order_status") \
        .agg(
            count("*").alias("order_count"),
            sum("amount").alias("total_revenue"),
            avg("amount").alias("avg_order_value"),
            countDistinct("customer_id").alias("unique_customers")
        )

    daily_metrics.write \
        .format("parquet") \
        .mode("overwrite") \
        .save(f"{gold_path}/daily_metrics")

    # Customer dimension
    customer_dim = silver_df \
        .groupBy("customer_id") \
        .agg(
            min("order_date").alias("first_order_date"),
            max("order_date").alias("last_order_date"),
            count("*").alias("total_orders"),
            sum("amount").alias("lifetime_value")
        ) \
        .withColumn("customer_tenure_days",
            datediff(current_date(), col("first_order_date"))
        ) \
        .withColumn("customer_segment",
            when(col("lifetime_value") > 10000, "enterprise")
            .when(col("lifetime_value") > 1000, "mid_market")
            .otherwise("smb")
        )

    customer_dim.write \
        .format("parquet") \
        .mode("overwrite") \
        .save(f"{gold_path}/dim_customers")

    print(f"Gold: Created daily_metrics and dim_customers")

# Data Quality Monitoring with Great Expectations
import great_expectations as gx
from great_expectations.core import ExpectationSuite

context = gx.get_context()

# Define quality expectations for Bronze layer
bronze_suite = ExpectationSuite(expectation_suite_name="bronze_orders")

bronze_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)
bronze_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
)
bronze_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="amount", min_value=0, max_value=1000000
    )
)
bronze_suite.add_expectation(
    gx.expectations.ExpectTableRowCountToBeBetween(
        min_value=1000, max_value=10000000
    )
)

# Run validation
result = context.run_checkpoint(
    checkpoint_name="bronze_checkpoint",
    batch_request={
        "datasource_name": "s3_datasource",
        "data_asset_name": "orders",
        "options": {"path": "s3://data-lake/bronze/orders/"},
    },
    suite=bronze_suite
)

print(f"Success: {result.success}")
print(f"Statistics: {result.statistics}")

# File format comparison and selection
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("FileFormats").getOrCreate()

# Read different formats
parquet_df = spark.read.format("parquet").load("s3://data-lake/events.parquet/")
orc_df = spark.read.format("orc").load("s3://data-lake/events.orc/")
json_df = spark.read.format("json").load("s3://data-lake/events.json/")

# Write optimized Parquet
parquet_df.write \
    .format("parquet") \
    .option("compression", "snappy") \
    .option("parquet.block.size", 128 * 1024 * 1024) \  # 128 MB blocks
    .partitionBy("event_date") \
    .save("s3://data-lake/optimized-events/")

# Schema evolution in Parquet
spark.read \
    .option("mergeSchema", "true") \
    .format("parquet") \
    .load("s3://data-lake/evolved-events/")

-- AWS Lake Formation: Grant table permissions
-- Grant SELECT on specific columns
GRANT SELECT ON TABLE analytics.events (event_id, event_type, event_date)
    TO IAM_ROLE 'arn:aws:iam::123456789:role/analyst-role';

-- Grant INSERT on staging schema
GRANT INSERT ON SCHEMA analytics.staging
    TO IAM_ROLE 'arn:aws:iam::123456789:role/etl-role';

-- Row-level security
CREATE ROW ACCESS POLICY events_policy ON analytics.events
    AS (event_type VARCHAR) RETURNS BOOLEAN ->
        CASE
            WHEN CURRENT_ROLE() = 'admin' THEN TRUE
            WHEN CURRENT_ROLE() = 'analyst' AND event_type IN ('click', 'view') THEN TRUE
            ELSE FALSE
        END;