Data Lake vs Data Warehouse

1-- Data Warehouse: Schema must be defined upfront
2CREATE TABLE customer_orders (
3    order_id INT PRIMARY KEY,
4    customer_id INT NOT NULL,
5    order_date DATE NOT NULL,
6    product_name VARCHAR(100),
7    quantity INT,
8    unit_price DECIMAL(10,2),
9    total_amount DECIMAL(10,2),
10    FOREIGN KEY (customer_id) REFERENCES customers(customer_id)
11);
12
13-- Data must conform to this schema during loading
14INSERT INTO customer_orders VALUES
15    (1001, 501, '2023-01-15', 'Laptop', 1, 1200.00, 1200.00);

1# Data Lake: Raw data stored as-is
2# File: s3://datalake/raw/orders/2023/01/15/orders.json
3
4{
5  "order_id": 1001,
6  "customer_id": 501,
7  "timestamp": "2023-01-15T10:30:00Z",
8  "items": [
9    {
10      "product": "Laptop",
11      "qty": 1,
12      "price": 1200.00
13    }
14  ],
15  "metadata": {
16    "source": "web",
17    "device": "mobile"
18  }
19}
20
21# Schema applied when reading with Apache Spark
22from pyspark.sql import SparkSession
23
24spark = SparkSession.builder.appName("DataLakeQuery").getOrCreate()
25
26# Read JSON with inferred schema
27df = spark.read.json("s3://datalake/raw/orders/2023/01/15/")
28
29# Or explicitly define schema at read time
30from pyspark.sql.types import StructType, StructField, IntegerType, StringType, ArrayType
31
32schema = StructType([
33    StructField("order_id", IntegerType(), True),
34    StructField("customer_id", IntegerType(), True),
35    StructField("timestamp", StringType(), True),
36    StructField("items", ArrayType(StructType([
37        StructField("product", StringType(), True),
38        StructField("qty", IntegerType(), True),
39        StructField("price", DoubleType(), True)
40    ])), True)
41])
42
43df = spark.read.schema(schema).json("s3://datalake/raw/orders/2023/01/15/")

1# Data Lake: Various data formats coexisting
2/datalake/
3  /raw/
4    /orders/          # Structured (Parquet)
5    /logs/            # Semi-structured (JSON)
6    /images/          # Unstructured (JPEG, PNG)
7    /videos/          # Unstructured (MP4)
8    /iot-sensors/     # Binary sensor data
9    /documents/       # Unstructured (PDF, DOCX)
10    /emails/          # Semi-structured (EML)
11    /social-media/    # Semi-structured (JSON)
12
13# Different tools can process different formats
14# Spark for structured/semi-structured
15df_orders = spark.read.parquet("s3://datalake/raw/orders/")
16df_logs = spark.read.json("s3://datalake/raw/logs/")
17
18# Python libraries for unstructured
19from PIL import Image
20image = Image.open("s3://datalake/raw/images/product_001.jpg")
21
22# Specialized ML libraries
23import cv2
24video = cv2.VideoCapture("s3://datalake/raw/videos/demo.mp4")

1-- ETL Example for Data Warehouse
2-- Step 1: Extract data from source
3CREATE TEMP TABLE staging_orders AS
4SELECT * FROM source_system.orders
5WHERE order_date >= CURRENT_DATE - 1;
6
7-- Step 2: Transform data (clean, standardize, aggregate)
8CREATE TEMP TABLE transformed_orders AS
9SELECT
10    order_id,
11    customer_id,
12    order_date,
13    UPPER(TRIM(customer_name)) AS customer_name,  -- Standardize
14    CASE
15        WHEN order_status = 'shipped' THEN 'Shipped'
16        WHEN order_status = 'delivered' THEN 'Delivered'
17        ELSE 'Processing'
18    END AS standardized_status,  -- Clean
19    total_amount,
20    CASE
21        WHEN total_amount > 1000 THEN 'High Value'
22        WHEN total_amount > 500 THEN 'Medium Value'
23        ELSE 'Low Value'
24    END AS value_category  -- Enrich
25FROM staging_orders
26WHERE total_amount > 0  -- Filter invalid data
27    AND customer_id IS NOT NULL;
28
29-- Step 3: Load into data warehouse
30INSERT INTO dim_customers (customer_id, customer_name)
31SELECT DISTINCT customer_id, customer_name
32FROM transformed_orders
33ON CONFLICT (customer_id) DO UPDATE
34    SET customer_name = EXCLUDED.customer_name;
35
36INSERT INTO fact_orders
37SELECT order_id, customer_id, order_date, standardized_status,
38       total_amount, value_category
39FROM transformed_orders;

1# ELT Example for Data Lake using PySpark
2
3from pyspark.sql import SparkSession
4from pyspark.sql.functions import *
5
6spark = SparkSession.builder.appName("ELT_Example").getOrCreate()
7
8# Step 1: Extract - Read data from source
9raw_orders = spark.read.format("jdbc") \\
10    .option("url", "jdbc:postgresql://source-db:5432/sales") \\
11    .option("dbtable", "orders") \\
12    .option("user", "reader") \\
13    .option("password", "secret") \\
14    .load()
15
16# Step 2: Load - Write raw data to data lake (Bronze layer)
17raw_orders.write.mode("append") \\
18    .partitionBy("order_date") \\
19    .parquet("s3://datalake/bronze/orders/")
20
21# Step 3: Transform - Applied later, as needed (Silver layer)
22# Different teams can transform the same data differently
23bronze_orders = spark.read.parquet("s3://datalake/bronze/orders/")
24
25# Marketing team's transformation
26marketing_view = bronze_orders \\
27    .filter(col("total_amount") > 0) \\
28    .withColumn("customer_name", upper(trim(col("customer_name")))) \\
29    .withColumn("value_segment",
30        when(col("total_amount") > 1000, "High Value")
31        .when(col("total_amount") > 500, "Medium Value")
32        .otherwise("Low Value")) \\
33    .select("order_id", "customer_id", "customer_name",
34            "order_date", "total_amount", "value_segment")
35
36marketing_view.write.mode("overwrite") \\
37    .partitionBy("order_date") \\
38    .parquet("s3://datalake/silver/marketing/customer_segments/")
39
40# Finance team's transformation (different view of same data)
41finance_view = bronze_orders \\
42    .filter(col("order_status").isin(["Shipped", "Delivered"])) \\
43    .groupBy("order_date", "order_status") \\
44    .agg(
45        sum("total_amount").alias("daily_revenue"),
46        count("order_id").alias("order_count")
47    )
48
49finance_view.write.mode("overwrite") \\
50    .partitionBy("order_date") \\
51    .parquet("s3://datalake/silver/finance/daily_revenue/")

1-- Data Warehouse: Highly optimized queries
2-- Complex analytical query runs in seconds
3SELECT
4    c.customer_segment,
5    c.region,
6    DATE_TRUNC('month', f.order_date) AS month,
7    COUNT(DISTINCT f.customer_id) AS unique_customers,
8    SUM(f.total_amount) AS total_revenue,
9    AVG(f.total_amount) AS avg_order_value,
10    SUM(f.total_amount) - LAG(SUM(f.total_amount)) OVER (
11        PARTITION BY c.customer_segment, c.region
12        ORDER BY DATE_TRUNC('month', f.order_date)
13    ) AS month_over_month_growth
14FROM fact_orders f
15JOIN dim_customers c ON f.customer_id = c.customer_id
16WHERE f.order_date >= '2023-01-01'
17GROUP BY c.customer_segment, c.region, month
18ORDER BY month DESC, total_revenue DESC;
19
20-- Benefits from:
21-- - Columnar storage (only reads needed columns)
22-- - Distribution keys (data co-located for joins)
23-- - Sort keys (efficient filtering and range scans)
24-- - Zone maps (skips irrelevant data blocks)

1# Data Lake: Query performance depends on organization and format
2from pyspark.sql import SparkSession
3from pyspark.sql.functions import *
4from pyspark.sql.window import Window
5
6spark = SparkSession.builder \\
7    .appName("DataLakeQuery") \\
8    .config("spark.sql.adaptive.enabled", "true") \\
9    .getOrCreate()
10
11# Reading well-organized Parquet files with partitioning
12orders = spark.read.parquet("s3://datalake/silver/orders/") \\
13    .filter(col("order_date") >= "2023-01-01")  # Partition pruning
14
15customers = spark.read.parquet("s3://datalake/silver/customers/")
16
17# Complex analytical query
18window_spec = Window.partitionBy("customer_segment", "region") \\
19    .orderBy("month")
20
21result = orders.join(customers, "customer_id") \\
22    .groupBy(
23        col("customer_segment"),
24        col("region"),
25        date_trunc("month", col("order_date")).alias("month")
26    ) \\
27    .agg(
28        countDistinct("customer_id").alias("unique_customers"),
29        sum("total_amount").alias("total_revenue"),
30        avg("total_amount").alias("avg_order_value")
31    ) \\
32    .withColumn("month_over_month_growth",
33        col("total_revenue") - lag("total_revenue").over(window_spec)
34    ) \\
35    .orderBy(desc("month"), desc("total_revenue"))
36
37# Performance optimizations:
38# - Use Parquet format (columnar, compressed)
39# - Partition data by frequently filtered columns
40# - Cache intermediate results if reused
41# - Use broadcast joins for small dimension tables
42# - Enable adaptive query execution

1-- Example: Redshift cost structure (simplified)
2-- Storage: $23/TB/month (compressed)
3-- Compute: Node-hour pricing
4-- RA3 nodes: $3.26/hour (16 TB SSD, 48 vCPUs, 384 GB RAM)
5
6-- Typical enterprise warehouse:
7-- Storage: 100 TB = $2,300/month
8-- Compute: 10 nodes × 24 hours × 30 days × $3.26 = $23,472/month
9-- Total: ~$26,000/month
10
11-- Cost optimization strategies:
12-- 1. Use compression (built-in)
13-- 2. Implement lifecycle policies
14-- 3. Use pause/resume for non-production
15-- 4. Optimize queries to reduce compute time
16-- 5. Right-size cluster based on workload

1# Example: AWS S3 + EMR cost structure (simplified)
2# Storage (S3):
3# - Standard: $0.023/GB/month ($23/TB/month)
4# - Infrequent Access: $0.0125/GB/month ($12.50/TB/month)
5# - Glacier: $0.004/GB/month ($4/TB/month)
6
7# Compute (EMR): Pay only when processing
8# - m5.xlarge: $0.192/hour (on-demand)
9# - m5.xlarge: $0.115/hour (spot instances, up to 40% savings)
10
11# Typical enterprise data lake:
12# Storage: 500 TB mixed tiers
13#   - 100 TB Standard (hot data): $2,300/month
14#   - 200 TB Infrequent Access: $2,500/month
15#   - 200 TB Glacier (archive): $800/month
16# Total storage: ~$5,600/month
17
18# Compute: Run jobs when needed
19# Daily ETL: 20 nodes × 4 hours × 30 days × $0.115 = $2,760/month
20# Ad-hoc analysis: 10 nodes × 40 hours/month × $0.192 = $768/month
21# Total compute: ~$3,528/month
22
23# Total: ~$9,128/month (65% cheaper than warehouse for this scenario)
24
25# Cost optimization strategies:
26# 1. Use appropriate storage tiers (S3 Intelligent-Tiering)
27# 2. Compress files (Parquet with Snappy)
28# 3. Use spot instances for non-critical workloads
29# 4. Partition data to reduce scan costs
30# 5. Implement lifecycle policies
31# 6. Use EMR auto-scaling
32# 7. Consider serverless options (AWS Athena) for infrequent queries

1# Delta Lake Example
2from pyspark.sql import SparkSession
3from delta import *
4
5# Configure Spark with Delta Lake
6builder = SparkSession.builder.appName("DeltaLakeExample") \\
7    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \\
8    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
9
10spark = configure_spark_with_delta_pip(builder).getOrCreate()
11
12# Create a Delta table with ACID properties
13data = spark.range(0, 5)
14data.write.format("delta").save("s3://datalake/delta/numbers")
15
16# Read Delta table
17df = spark.read.format("delta").load("s3://datalake/delta/numbers")
18
19# ACID transactions: Update, Delete, Merge operations
20from delta.tables import *
21
22deltaTable = DeltaTable.forPath(spark, "s3://datalake/delta/numbers")
23
24# Update operation (ACID)
25deltaTable.update(
26    condition = "id % 2 == 0",
27    set = { "id": "id + 100" }
28)
29
30# Delete operation (ACID)
31deltaTable.delete(condition = "id < 2")
32
33# Merge operation (Upsert) - ACID
34newData = spark.range(0, 3)
35
36deltaTable.alias("target").merge(
37    newData.alias("source"),
38    "target.id = source.id"
39).whenMatchedUpdate(set = { "id": "source.id + 1000" }) \\
40 .whenNotMatchedInsert(values = { "id": "source.id" }) \\
41 .execute()
42
43# Time Travel: Query historical versions
44df_v0 = spark.read.format("delta").option("versionAsOf", 0).load("s3://datalake/delta/numbers")
45df_yesterday = spark.read.format("delta") \\
46    .option("timestampAsOf", "2023-01-15") \\
47    .load("s3://datalake/delta/numbers")
48
49# Schema Evolution: Automatically handle schema changes
50newData = spark.createDataFrame([(6, "new_column")], ["id", "new_col"])
51newData.write.format("delta") \\
52    .mode("append") \\
53    .option("mergeSchema", "true") \\
54    .save("s3://datalake/delta/numbers")
55
56# Z-Ordering for query optimization
57spark.sql("OPTIMIZE delta.\`s3://datalake/delta/numbers\` ZORDER BY (id)")
58
59# Vacuum: Remove old files no longer needed
60deltaTable.vacuum(retentionHours=168)  # Remove files older than 7 days

1# Apache Iceberg Example
2from pyspark.sql import SparkSession
3
4# Configure Spark with Iceberg
5spark = SparkSession.builder \\
6    .appName("IcebergExample") \\
7    .config("spark.sql.catalog.my_catalog", "org.apache.iceberg.spark.SparkCatalog") \\
8    .config("spark.sql.catalog.my_catalog.type", "hadoop") \\
9    .config("spark.sql.catalog.my_catalog.warehouse", "s3://datalake/iceberg/") \\
10    .getOrCreate()
11
12# Create Iceberg table
13spark.sql("""
14    CREATE TABLE my_catalog.db.orders (
15        order_id BIGINT,
16        customer_id BIGINT,
17        order_date DATE,
18        total_amount DECIMAL(10,2)
19    ) USING iceberg
20    PARTITIONED BY (days(order_date))
21""")
22
23# Insert data
24spark.sql("""
25    INSERT INTO my_catalog.db.orders VALUES
26    (1, 101, DATE '2023-01-15', 1200.00),
27    (2, 102, DATE '2023-01-16', 800.00)
28""")
29
30# Hidden partitioning: Users don't need to specify partition columns
31spark.sql("""
32    SELECT * FROM my_catalog.db.orders
33    WHERE order_date BETWEEN DATE '2023-01-01' AND DATE '2023-01-31'
34""")  # Automatically uses partition pruning
35
36# Time Travel: Query historical snapshots
37spark.sql("""
38    SELECT * FROM my_catalog.db.orders
39    VERSION AS OF 1234567890
40""")
41
42spark.sql("""
43    SELECT * FROM my_catalog.db.orders
44    TIMESTAMP AS OF '2023-01-15 10:00:00'
45""")
46
47# Schema Evolution
48spark.sql("""
49    ALTER TABLE my_catalog.db.orders
50    ADD COLUMN order_status STRING
51""")
52
53# Merge operation (Upsert)
54spark.sql("""
55    MERGE INTO my_catalog.db.orders t
56    USING (SELECT 1 as order_id, 101 as customer_id,
57           DATE '2023-01-15' as order_date, 1500.00 as total_amount) s
58    ON t.order_id = s.order_id
59    WHEN MATCHED THEN UPDATE SET t.total_amount = s.total_amount
60    WHEN NOT MATCHED THEN INSERT *
61""")
62
63# Compaction: Optimize small files
64spark.sql("CALL my_catalog.system.rewrite_data_files('db.orders')")
65
66# Remove old snapshots
67spark.sql("CALL my_catalog.system.expire_snapshots('db.orders', TIMESTAMP '2023-01-01 00:00:00')")

1# Apache Hudi Example
2from pyspark.sql import SparkSession
3
4spark = SparkSession.builder \\
5    .appName("HudiExample") \\
6    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \\
7    .getOrCreate()
8
9# Hudi configuration
10hudi_options = {
11    'hoodie.table.name': 'orders',
12    'hoodie.datasource.write.recordkey.field': 'order_id',
13    'hoodie.datasource.write.precombine.field': 'updated_at',
14    'hoodie.datasource.write.partitionpath.field': 'order_date',
15    'hoodie.datasource.write.table.type': 'COPY_ON_WRITE',  # or MERGE_ON_READ
16    'hoodie.datasource.write.operation': 'upsert'
17}
18
19# Initial write
20orders_df = spark.read.parquet("s3://source/orders/")
21
22orders_df.write.format("hudi") \\
23    .options(**hudi_options) \\
24    .mode("overwrite") \\
25    .save("s3://datalake/hudi/orders/")
26
27# Incremental upsert (update existing, insert new)
28new_orders_df = spark.read.parquet("s3://source/orders/incremental/")
29
30new_orders_df.write.format("hudi") \\
31    .options(**hudi_options) \\
32    .mode("append") \\
33    .save("s3://datalake/hudi/orders/")
34
35# Read Hudi table
36hudi_df = spark.read.format("hudi").load("s3://datalake/hudi/orders/")
37
38# Incremental query: Read only changes since a specific commit
39incremental_df = spark.read.format("hudi") \\
40    .option("hoodie.datasource.query.type", "incremental") \\
41    .option("hoodie.datasource.read.begin.instanttime", "20230115000000") \\
42    .load("s3://datalake/hudi/orders/")
43
44# Time Travel query
45snapshot_df = spark.read.format("hudi") \\
46    .option("as.of.instant", "20230115100000") \\
47    .load("s3://datalake/hudi/orders/")
48
49# Delete records
50delete_keys = spark.read.json("s3://source/delete_keys/")
51
52hudi_delete_options = {
53    **hudi_options,
54    'hoodie.datasource.write.operation': 'delete'
55}
56
57delete_keys.write.format("hudi") \\
58    .options(**hudi_delete_options) \\
59    .mode("append") \\
60    .save("s3://datalake/hudi/orders/")
61
62# Compaction for MERGE_ON_READ tables
63spark.read.format("hudi") \\
64    .option("hoodie.compact.inline", "true") \\
65    .load("s3://datalake/hudi/orders/")

1# Bronze Layer: Ingest raw data
2from pyspark.sql import SparkSession
3from pyspark.sql.functions import current_timestamp, input_file_name
4
5spark = SparkSession.builder.appName("Bronze Layer").getOrCreate()
6
7# Read from source (example: streaming from Kafka)
8raw_stream = spark.readStream \\
9    .format("kafka") \\
10    .option("kafka.bootstrap.servers", "localhost:9092") \\
11    .option("subscribe", "orders") \\
12    .load()
13
14# Add metadata
15bronze_df = raw_stream.selectExpr("CAST(value AS STRING) as json_data") \\
16    .withColumn("ingestion_timestamp", current_timestamp()) \\
17    .withColumn("source", lit("kafka_orders"))
18
19# Write to Bronze layer (append-only, immutable)
20bronze_df.writeStream \\
21    .format("delta") \\
22    .outputMode("append") \\
23    .option("checkpointLocation", "s3://datalake/checkpoints/bronze/orders") \\
24    .partitionBy("date") \\
25    .start("s3://datalake/bronze/orders/")
26
27# Characteristics of Bronze layer:
28# - Exact copy of source data
29# - Append-only (no updates or deletes)
30# - Include ingestion metadata
31# - Partitioned by ingestion date
32# - Used for audit trail and reprocessing

1# Silver Layer: Clean and validate data
2from pyspark.sql.functions import *
3from pyspark.sql.types import *
4
5# Read from Bronze
6bronze_df = spark.readStream \\
7    .format("delta") \\
8    .load("s3://datalake/bronze/orders/")
9
10# Define schema for parsing JSON
11order_schema = StructType([
12    StructField("order_id", LongType(), False),
13    StructField("customer_id", LongType(), False),
14    StructField("order_date", StringType(), False),
15    StructField("total_amount", DoubleType(), True),
16    StructField("status", StringType(), True),
17    StructField("items", ArrayType(StructType([
18        StructField("product_id", LongType()),
19        StructField("quantity", IntegerType()),
20        StructField("price", DoubleType())
21    ])))
22])
23
24# Parse JSON and clean data
25silver_df = bronze_df \\
26    .select(from_json(col("json_data"), order_schema).alias("data")) \\
27    .select("data.*") \\
28    .filter(col("order_id").isNotNull()) \\  # Data quality: remove nulls
29    .filter(col("total_amount") > 0) \\      # Data quality: valid amounts
30    .withColumn("order_date", to_date(col("order_date"))) \\  # Standardize format
31    .withColumn("status", upper(trim(col("status")))) \\      # Standardize values
32    .withColumn("is_valid",
33        when((col("total_amount") > 0) &
34             (col("customer_id").isNotNull()) &
35             (col("order_date").isNotNull()), True)
36        .otherwise(False)
37    ) \\
38    .dropDuplicates(["order_id"]) \\  # Remove duplicates
39    .withColumn("processed_timestamp", current_timestamp())
40
41# Write to Silver layer
42silver_df.writeStream \\
43    .format("delta") \\
44    .outputMode("append") \\
45    .option("checkpointLocation", "s3://datalake/checkpoints/silver/orders") \\
46    .partitionBy("order_date") \\
47    .start("s3://datalake/silver/orders/")
48
49# Characteristics of Silver layer:
50# - Cleaned and validated data
51# - Standardized formats and values
52# - Schema enforcement
53# - Deduplication
54# - Data quality flags
55# - Can support updates (SCD Type 1 or 2)
56# - Used by data scientists and analysts

1# Gold Layer: Create business-ready aggregates
2from pyspark.sql.window import Window
3
4# Read from Silver
5silver_orders = spark.read.format("delta").load("s3://datalake/silver/orders/")
6silver_customers = spark.read.format("delta").load("s3://datalake/silver/customers/")
7
8# Create Gold table: Daily sales summary
9daily_sales = silver_orders \\
10    .groupBy("order_date") \\
11    .agg(
12        count("order_id").alias("total_orders"),
13        sum("total_amount").alias("total_revenue"),
14        avg("total_amount").alias("avg_order_value"),
15        countDistinct("customer_id").alias("unique_customers")
16    ) \\
17    .orderBy("order_date")
18
19# Write Gold table
20daily_sales.write \\
21    .format("delta") \\
22    .mode("overwrite") \\
23    .option("overwriteSchema", "true") \\
24    .save("s3://datalake/gold/sales/daily_summary/")
25
26# Create Gold table: Customer 360 view
27window_spec = Window.partitionBy("customer_id").orderBy(desc("order_date"))
28
29customer_360 = silver_orders \\
30    .join(silver_customers, "customer_id") \\
31    .withColumn("latest_order", first("order_date").over(window_spec)) \\
32    .withColumn("lifetime_orders", count("order_id").over(Window.partitionBy("customer_id"))) \\
33    .withColumn("lifetime_value", sum("total_amount").over(Window.partitionBy("customer_id"))) \\
34    .withColumn("avg_order_value", avg("total_amount").over(Window.partitionBy("customer_id"))) \\
35    .select(
36        "customer_id",
37        "customer_name",
38        "customer_segment",
39        "latest_order",
40        "lifetime_orders",
41        "lifetime_value",
42        "avg_order_value"
43    ) \\
44    .dropDuplicates(["customer_id"])
45
46customer_360.write \\
47    .format("delta") \\
48    .mode("overwrite") \\
49    .save("s3://datalake/gold/customers/customer_360/")
50
51# Create Gold table: Product performance
52product_performance = silver_orders \\
53    .select(explode("items").alias("item")) \\
54    .select(
55        "item.product_id",
56        "item.quantity",
57        "item.price",
58        col("order_date")
59    ) \\
60    .groupBy("product_id", date_trunc("month", "order_date").alias("month")) \\
61    .agg(
62        sum("quantity").alias("units_sold"),
63        sum(col("quantity") * col("price")).alias("revenue"),
64        count("*").alias("orders_with_product")
65    )
66
67product_performance.write \\
68    .format("delta") \\
69    .mode("overwrite") \\
70    .partitionBy("month") \\
71    .save("s3://datalake/gold/products/monthly_performance/")
72
73# Characteristics of Gold layer:
74# - Business-level aggregates
75# - Optimized for specific queries
76# - Feature engineering for ML
77# - Curated datasets for BI tools
78# - May include calculated metrics and KPIs
79# - Read-optimized (often fully denormalized)
80# - Used by business users and applications