AWS Lakehouse Architecture: Hướng Dẫn Toàn Diện Xây Dựng Data Platform Hiện Đại

┌─────────────────────────────────────────────────────────────┐
│                    Query & Analytics Layer                  │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐       │
│  │   Athena     │  │  Spark/EMR   │  │  QuickSight  │       │
│  └──────────────┘  └──────────────┘  └──────────────┘       │
└─────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────┐
│              Metadata & Catalog Layer                       │
│  ┌──────────────────────────────────────────────────┐       │
│  │         AWS Glue Data Catalog                    │       │
│  │  (Table registrations, Schema, Partitions)       │       │
│  └──────────────────────────────────────────────────┘       │
└─────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────┐
│                  Table Format Layer                         │
│  ┌──────────────────────────────────────────────────┐       │
│  │            Apache Iceberg                        │       │
│  │  (Snapshots, Manifests, Partition Metadata)      │       │
│  └──────────────────────────────────────────────────┘       │
└─────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────┐
│                    Storage Layer                            │
│  ┌──────────────────────────────────────────────────┐       │
│  │              Amazon S3                           │       │
│  │  (Data files: Parquet/ORC + Iceberg metadata)    │       │
│  └──────────────────────────────────────────────────┘       │
└─────────────────────────────────────────────────────────────┘

Table Metadata File (JSON)
    ↓
Snapshot (version của table tại một thời điểm)
    ↓
Manifest List (danh sách các manifest files)
    ↓
Manifest Files (metadata về data files)
    ↓
Data Files (Parquet/ORC files chứa actual data)

Source Systems → Kinesis/Kafka → S3 (Raw Zone)
                    ↓
              Glue ETL Job
                    ↓
         S3 (Processed Zone - Iceberg Tables)

Raw Iceberg Tables → Glue/Spark ETL → Curated Iceberg Tables
                                            ↓
                                    Glue Catalog Registration

User/Application → Athena → Glue Catalog → Iceberg Metadata → S3 Data Files

my-lakehouse-bucket/
├── raw/              # Bronze zone - raw ingested data
│   ├── source1/
│   └── source2/
├── processed/        # Silver zone - cleaned and validated data
│   ├── table1/
│   └── table2/
└── curated/          # Gold zone - business-ready aggregated data
    ├── analytics/
    └── reporting/

# main.tf
terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
}

provider "aws" {
  region = var.aws_region
}

# S3 Bucket cho Lakehouse
resource "aws_s3_bucket" "lakehouse" {
  bucket = "${var.project_name}-lakehouse-${var.environment}"

  tags = {
    Name        = "Lakehouse Data Storage"
    Environment = var.environment
    ManagedBy   = "Terraform"
  }
}

# Enable versioning
resource "aws_s3_bucket_versioning" "lakehouse" {
  bucket = aws_s3_bucket.lakehouse.id

  versioning_configuration {
    status = "Enabled"
  }
}

# Server-side encryption
resource "aws_s3_bucket_server_side_encryption_configuration" "lakehouse" {
  bucket = aws_s3_bucket.lakehouse.id

  rule {
    apply_server_side_encryption_by_default {
      sse_algorithm = "AES256"
    }
  }
}

# Block public access
resource "aws_s3_bucket_public_access_block" "lakehouse" {
  bucket = aws_s3_bucket.lakehouse.id

  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

# Lifecycle policy cho cost optimization
resource "aws_s3_bucket_lifecycle_configuration" "lakehouse" {
  bucket = aws_s3_bucket.lakehouse.id

  rule {
    id     = "transition-raw-to-ia"
    status = "Enabled"

    filter {
      prefix = "raw/"
    }

    transition {
      days          = 30
      storage_class = "STANDARD_IA"
    }

    transition {
      days          = 90
      storage_class = "GLACIER_IR"
    }
  }

  rule {
    id     = "expire-old-versions"
    status = "Enabled"

    noncurrent_version_expiration {
      noncurrent_days = 90
    }
  }
}

# Variables
variable "project_name" {
  description = "Project name for resource naming"
  type        = string
  default     = "my-lakehouse"
}

variable "environment" {
  description = "Environment (dev, staging, prod)"
  type        = string
  default     = "dev"
}

variable "aws_region" {
  description = "AWS region"
  type        = string
  default     = "us-east-1"
}

# s3-lakehouse.yaml
AWSTemplateFormatVersion: '2010-09-09'
Description: 'S3 Bucket for Data Lakehouse'

Parameters:
  ProjectName:
    Type: String
    Default: my-lakehouse
    Description: Project name for resource naming

  Environment:
    Type: String
    Default: dev
    AllowedValues:
      - dev
      - staging
      - prod
    Description: Environment name

Resources:
  LakehouseBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub '${ProjectName}-lakehouse-${Environment}'
      VersioningConfiguration:
        Status: Enabled
      BucketEncryption:
        ServerSideEncryptionConfiguration:
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: AES256
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true
      LifecycleConfiguration:
        Rules:
          - Id: TransitionRawToIA
            Status: Enabled
            Prefix: raw/
            Transitions:
              - TransitionInDays: 30
                StorageClass: STANDARD_IA
              - TransitionInDays: 90
                StorageClass: GLACIER_IR
          - Id: ExpireOldVersions
            Status: Enabled
            NoncurrentVersionExpiration:
              NoncurrentDays: 90
      Tags:
        - Key: Name
          Value: Lakehouse Data Storage
        - Key: Environment
          Value: !Ref Environment
        - Key: ManagedBy
          Value: CloudFormation

Outputs:
  BucketName:
    Description: Name of the S3 bucket
    Value: !Ref LakehouseBucket
    Export:
      Name: !Sub '${AWS::StackName}-BucketName'

  BucketArn:
    Description: ARN of the S3 bucket
    Value: !GetAtt LakehouseBucket.Arn
    Export:
      Name: !Sub '${AWS::StackName}-BucketArn'

# create_glue_database.py
import boto3

glue_client = boto3.client('glue', region_name='us-east-1')

def create_glue_database(database_name, description):
    """
    Tạo Glue database cho lakehouse
    """
    try:
        response = glue_client.create_database(
            DatabaseInput={
                'Name': database_name,
                'Description': description,
                'LocationUri': f's3://my-lakehouse-bucket/processed/',
                'Parameters': {
                    'classification': 'iceberg',
                    'table_type': 'ICEBERG'
                }
            }
        )
        print(f"Database {database_name} created successfully")
        return response
    except glue_client.exceptions.AlreadyExistsException:
        print(f"Database {database_name} already exists")
    except Exception as e:
        print(f"Error creating database: {str(e)}")
        raise

# Tạo databases cho các zones khác nhau
create_glue_database('lakehouse_raw', 'Raw data zone')
create_glue_database('lakehouse_processed', 'Processed data zone')
create_glue_database('lakehouse_curated', 'Curated data zone')

# glue.tf
resource "aws_glue_catalog_database" "raw" {
  name        = "${var.project_name}_raw"
  description = "Raw data zone"

  location_uri = "s3://${aws_s3_bucket.lakehouse.id}/raw/"

  parameters = {
    classification = "iceberg"
    table_type     = "ICEBERG"
  }
}

resource "aws_glue_catalog_database" "processed" {
  name        = "${var.project_name}_processed"
  description = "Processed data zone"

  location_uri = "s3://${aws_s3_bucket.lakehouse.id}/processed/"

  parameters = {
    classification = "iceberg"
    table_type     = "ICEBERG"
  }
}

resource "aws_glue_catalog_database" "curated" {
  name        = "${var.project_name}_curated"
  description = "Curated data zone"

  location_uri = "s3://${aws_s3_bucket.lakehouse.id}/curated/"

  parameters = {
    classification = "iceberg"
    table_type     = "ICEBERG"
  }
}

-- create_iceberg_table.sql
-- Tạo Iceberg table trong Athena

CREATE TABLE lakehouse_processed.customer_events (
    event_id STRING,
    customer_id STRING,
    event_type STRING,
    event_timestamp TIMESTAMP,
    event_data MAP<STRING, STRING>,
    event_date DATE
)
PARTITIONED BY (event_date)
LOCATION 's3://my-lakehouse-bucket/processed/customer_events/'
TBLPROPERTIES (
    'table_type' = 'ICEBERG',
    'format' = 'parquet',
    'write_compression' = 'snappy',
    'optimize_rewrite_delete_file_threshold' = '10'
);

# create_iceberg_table_spark.py
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, TimestampType, DateType, MapType

# Khởi tạo Spark session với Iceberg support
spark = SparkSession.builder \
    .appName("CreateIcebergTable") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.glue_catalog", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.glue_catalog.warehouse", "s3://my-lakehouse-bucket/processed/") \
    .config("spark.sql.catalog.glue_catalog.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog") \
    .config("spark.sql.catalog.glue_catalog.io-impl", "org.apache.iceberg.aws.s3.S3FileIO") \
    .getOrCreate()

# Define schema
schema = StructType([
    StructField("event_id", StringType(), False),
    StructField("customer_id", StringType(), False),
    StructField("event_type", StringType(), False),
    StructField("event_timestamp", TimestampType(), False),
    StructField("event_data", MapType(StringType(), StringType()), True),
    StructField("event_date", DateType(), False)
])

# Tạo Iceberg table
spark.sql("""
    CREATE TABLE IF NOT EXISTS glue_catalog.lakehouse_processed.customer_events (
        event_id STRING,
        customer_id STRING,
        event_type STRING,
        event_timestamp TIMESTAMP,
        event_data MAP<STRING, STRING>,
        event_date DATE
    )
    USING iceberg
    PARTITIONED BY (days(event_date))
    LOCATION 's3://my-lakehouse-bucket/processed/customer_events/'
    TBLPROPERTIES (
        'format-version' = '2',
        'write.parquet.compression-codec' = 'snappy'
    )
""")

print("Iceberg table created successfully")

# glue_etl_iceberg.py
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.dynamicframe import DynamicFrame
from pyspark.sql.functions import col, to_date, current_timestamp

# Initialize contexts
args = getResolvedOptions(sys.argv, ['JOB_NAME', 'database_name', 'table_name', 'source_path'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Configure Iceberg
spark.conf.set("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
spark.conf.set("spark.sql.catalog.glue_catalog", "org.apache.iceberg.spark.SparkCatalog")
spark.conf.set("spark.sql.catalog.glue_catalog.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog")
spark.conf.set("spark.sql.catalog.glue_catalog.io-impl", "org.apache.iceberg.aws.s3.S3FileIO")

# Read source data
source_df = spark.read \
    .format("json") \
    .load(args['source_path'])

# Transform data
transformed_df = source_df \
    .withColumn("event_date", to_date(col("event_timestamp"))) \
    .withColumn("processed_at", current_timestamp())

# Write to Iceberg table
transformed_df.writeTo(f"glue_catalog.{args['database_name']}.{args['table_name']}") \
    .using("iceberg") \
    .tableProperty("format-version", "2") \
    .tableProperty("write.parquet.compression-codec", "snappy") \
    .append()

job.commit()
print(f"Successfully wrote data to {args['database_name']}.{args['table_name']}")

# glue_job.tf
resource "aws_iam_role" "glue_job_role" {
  name = "${var.project_name}-glue-job-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "glue.amazonaws.com"
        }
      }
    ]
  })
}

resource "aws_iam_role_policy_attachment" "glue_service_role" {
  role       = aws_iam_role.glue_job_role.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole"
}

resource "aws_iam_role_policy" "glue_s3_policy" {
  name = "${var.project_name}-glue-s3-policy"
  role = aws_iam_role.glue_job_role.id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "s3:GetObject",
          "s3:PutObject",
          "s3:DeleteObject",
          "s3:ListBucket"
        ]
        Resource = [
          "${aws_s3_bucket.lakehouse.arn}",
          "${aws_s3_bucket.lakehouse.arn}/*"
        ]
      }
    ]
  })
}

resource "aws_glue_job" "iceberg_etl" {
  name     = "${var.project_name}-iceberg-etl"
  role_arn = aws_iam_role.glue_job_role.arn

  command {
    name            = "glueetl"
    script_location = "s3://${aws_s3_bucket.lakehouse.id}/scripts/glue_etl_iceberg.py"
    python_version  = "3"
  }

  default_arguments = {
    "--job-language"                     = "python"
    "--job-bookmark-option"              = "job-bookmark-enable"
    "--enable-metrics"                   = "true"
    "--enable-continuous-cloudwatch-log" = "true"
    "--enable-spark-ui"                  = "true"
    "--spark-event-logs-path"            = "s3://${aws_s3_bucket.lakehouse.id}/spark-logs/"
    "--datalake-formats"                 = "iceberg"
    "--conf"                             = "spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"
  }

  glue_version      = "4.0"
  max_retries       = 1
  timeout           = 60
  number_of_workers = 10
  worker_type       = "G.1X"

  tags = {
    Name        = "Iceberg ETL Job"
    Environment = var.environment
  }
}

-- Query Iceberg table
SELECT
    event_type,
    COUNT(*) as event_count,
    COUNT(DISTINCT customer_id) as unique_customers
FROM lakehouse_processed.customer_events
WHERE event_date >= DATE '2024-01-01'
GROUP BY event_type
ORDER BY event_count DESC;

-- Time travel query (query snapshot tại thời điểm cụ thể)
SELECT *
FROM lakehouse_processed.customer_events
FOR SYSTEM_TIME AS OF TIMESTAMP '2024-01-15 10:00:00'
WHERE customer_id = 'CUST123';

-- Query specific snapshot
SELECT *
FROM lakehouse_processed.customer_events
FOR SYSTEM_VERSION AS OF 1234567890
LIMIT 100;

# query_athena.py
import boto3
import time
import pandas as pd

class AthenaQueryExecutor:
    def __init__(self, database, output_location, region='us-east-1'):
        self.athena_client = boto3.client('athena', region_name=region)
        self.s3_client = boto3.client('s3', region_name=region)
        self.database = database
        self.output_location = output_location

    def execute_query(self, query, wait=True):
        """
        Execute Athena query và return results
        """
        response = self.athena_client.start_query_execution(
            QueryString=query,
            QueryExecutionContext={'Database': self.database},
            ResultConfiguration={'OutputLocation': self.output_location}
        )

        query_execution_id = response['QueryExecutionId']
        print(f"Query execution ID: {query_execution_id}")

        if wait:
            return self._wait_for_query_completion(query_execution_id)

        return query_execution_id

    def _wait_for_query_completion(self, query_execution_id):
        """
        Wait cho query completion và return results
        """
        while True:
            response = self.athena_client.get_query_execution(
                QueryExecutionId=query_execution_id
            )

            status = response['QueryExecution']['Status']['State']

            if status in ['SUCCEEDED', 'FAILED', 'CANCELLED']:
                break

            print(f"Query status: {status}")
            time.sleep(2)

        if status == 'SUCCEEDED':
            print("Query succeeded!")
            return self._get_query_results(query_execution_id)
        else:
            error_message = response['QueryExecution']['Status'].get('StateChangeReason', 'Unknown error')
            raise Exception(f"Query failed with status {status}: {error_message}")

    def _get_query_results(self, query_execution_id):
        """
        Get query results và convert to pandas DataFrame
        """
        results = self.athena_client.get_query_results(
            QueryExecutionId=query_execution_id
        )

        # Extract column names
        columns = [col['Label'] for col in results['ResultSet']['ResultSetMetadata']['ColumnInfo']]

        # Extract rows
        rows = []
        for row in results['ResultSet']['Rows'][1:]:  # Skip header row
            rows.append([field.get('VarCharValue', '') for field in row['Data']])

        # Create DataFrame
        df = pd.DataFrame(rows, columns=columns)
        return df

# Usage example
if __name__ == "__main__":
    executor = AthenaQueryExecutor(
        database='lakehouse_processed',
        output_location='s3://my-lakehouse-bucket/athena-results/'
    )

    query = """
    SELECT
        event_type,
        COUNT(*) as event_count
    FROM customer_events
    WHERE event_date >= DATE '2024-01-01'
    GROUP BY event_type
    """

    results_df = executor.execute_query(query)
    print(results_df)

# Register Iceberg table trong Glue Catalog
glue_client.create_table(
    DatabaseName='lakehouse_processed',
    TableInput={
        'Name': 'customer_events',
        'StorageDescriptor': {
            'Location': 's3://my-lakehouse-bucket/processed/customer_events/',
            'InputFormat': 'org.apache.iceberg.mr.mapreduce.IcebergInputFormat',
            'OutputFormat': 'org.apache.iceberg.mr.mapreduce.IcebergOutputFormat',
            'SerdeInfo': {
                'SerializationLibrary': 'org.apache.iceberg.mr.hive.HiveIcebergSerDe'
            }
        },
        'Parameters': {
            'table_type': 'ICEBERG',
            'metadata_location': 's3://my-lakehouse-bucket/processed/customer_events/metadata/v1.metadata.json'
        }
    }
)

-- Athena SQL để expire snapshots cũ hơn 7 ngày
CALL system.expire_snapshots(
    table => 'lakehouse_processed.customer_events',
    older_than => TIMESTAMP '2024-01-01 00:00:00',
    retain_last => 5
);

-- Remove files không còn được reference bởi bất kỳ snapshot nào
CALL system.remove_orphan_files(
    table => 'lakehouse_processed.customer_events',
    older_than => TIMESTAMP '2024-01-01 00:00:00'
);

# PySpark code để compact manifests
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .config("spark.sql.catalog.glue_catalog", "org.apache.iceberg.spark.SparkCatalog") \
    .getOrCreate()

# Rewrite manifests để reduce metadata overhead
spark.sql("""
    CALL glue_catalog.system.rewrite_manifests(
        'lakehouse_processed.customer_events'
    )
""")

# glue_maintenance_job.py
import boto3
from datetime import datetime, timedelta

def run_maintenance_tasks(database, table):
    """
    Run automated maintenance tasks cho Iceberg table
    """
    athena_client = boto3.client('athena')

    # Calculate retention date (7 days ago)
    retention_date = (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d %H:%M:%S')

    tasks = [
        # Expire old snapshots
        f"""
        CALL system.expire_snapshots(
            table => '{database}.{table}',
            older_than => TIMESTAMP '{retention_date}',
            retain_last => 5
        )
        """,

        # Remove orphan files
        f"""
        CALL system.remove_orphan_files(
            table => '{database}.{table}',
            older_than => TIMESTAMP '{retention_date}'
        )
        """,

        # Rewrite data files (compact small files)
        f"""
        CALL system.rewrite_data_files(
            table => '{database}.{table}',
            options => map('target-file-size-bytes', '536870912')
        )
        """
    ]

    for task in tasks:
        print(f"Executing: {task}")
        response = athena_client.start_query_execution(
            QueryString=task,
            QueryExecutionContext={'Database': database},
            ResultConfiguration={
                'OutputLocation': 's3://my-lakehouse-bucket/athena-results/'
            }
        )
        print(f"Query execution ID: {response['QueryExecutionId']}")

# Schedule này có thể được run bởi EventBridge hoặc Glue workflow
if __name__ == "__main__":
    run_maintenance_tasks('lakehouse_processed', 'customer_events')

-- Query để check snapshot count
SELECT
    COUNT(*) as snapshot_count,
    MIN(committed_at) as oldest_snapshot,
    MAX(committed_at) as newest_snapshot
FROM lakehouse_processed.customer_events.snapshots;

-- Query để check manifest file count
SELECT
    snapshot_id,
    COUNT(*) as manifest_count
FROM lakehouse_processed.customer_events.manifests
GROUP BY snapshot_id
ORDER BY manifest_count DESC
LIMIT 10;

-- Query để analyze file sizes
SELECT
    COUNT(*) as file_count,
    AVG(file_size_in_bytes) / 1024 / 1024 as avg_file_size_mb,
    MIN(file_size_in_bytes) / 1024 / 1024 as min_file_size_mb,
    MAX(file_size_in_bytes) / 1024 / 1024 as max_file_size_mb
FROM lakehouse_processed.customer_events.files;

# publish_metadata_metrics.py
import boto3
from pyspark.sql import SparkSession

def publish_metadata_metrics(database, table):
    """
    Collect và publish metadata metrics to CloudWatch
    """
    cloudwatch = boto3.client('cloudwatch')
    spark = SparkSession.builder.getOrCreate()

    # Get snapshot count
    snapshot_df = spark.sql(f"SELECT COUNT(*) as count FROM {database}.{table}.snapshots")
    snapshot_count = snapshot_df.collect()[0]['count']

    # Get file count và average size
    files_df = spark.sql(f"""
        SELECT
            COUNT(*) as file_count,
            AVG(file_size_in_bytes) as avg_file_size
        FROM {database}.{table}.files
    """)
    file_stats = files_df.collect()[0]

    # Publish metrics
    cloudwatch.put_metric_data(
        Namespace='Lakehouse/Metadata',
        MetricData=[
            {
                'MetricName': 'SnapshotCount',
                'Value': snapshot_count,
                'Unit': 'Count',
                'Dimensions': [
                    {'Name': 'Database', 'Value': database},
                    {'Name': 'Table', 'Value': table}
                ]
            },
            {
                'MetricName': 'DataFileCount',
                'Value': file_stats['file_count'],
                'Unit': 'Count',
                'Dimensions': [
                    {'Name': 'Database', 'Value': database},
                    {'Name': 'Table', 'Value': table}
                ]
            },
            {
                'MetricName': 'AverageFileSize',
                'Value': file_stats['avg_file_size'] / 1024 / 1024,  # Convert to MB
                'Unit': 'Megabytes',
                'Dimensions': [
                    {'Name': 'Database', 'Value': database},
                    {'Name': 'Table', 'Value': table}
                ]
            }
        ]
    )

    print(f"Published metrics for {database}.{table}")

# Terraform configuration đã include versioning trong section 3.2.2
resource "aws_s3_bucket_versioning" "lakehouse" {
  bucket = aws_s3_bucket.lakehouse.id

  versioning_configuration {
    status = "Enabled"
  }
}

# backup_metadata.py
import boto3
from datetime import datetime

def backup_table_metadata(source_bucket, table_path, backup_bucket):
    """
    Backup Iceberg table metadata to separate bucket
    """
    s3_client = boto3.client('s3')

    # List all metadata files
    metadata_prefix = f"{table_path}/metadata/"
    response = s3_client.list_objects_v2(
        Bucket=source_bucket,
        Prefix=metadata_prefix
    )

    if 'Contents' not in response:
        print(f"No metadata files found at {metadata_prefix}")
        return

    # Backup each metadata file
    backup_timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')

    for obj in response['Contents']:
        source_key = obj['Key']
        backup_key = f"backups/{backup_timestamp}/{source_key}"

        # Copy to backup bucket
        s3_client.copy_object(
            CopySource={'Bucket': source_bucket, 'Key': source_key},
            Bucket=backup_bucket,
            Key=backup_key
        )

        print(f"Backed up {source_key} to {backup_key}")

# Usage
backup_table_metadata(
    source_bucket='my-lakehouse-bucket',
    table_path='processed/customer_events',
    backup_bucket='my-lakehouse-backup-bucket'
)

-- Athena SQL để add column
ALTER TABLE lakehouse_processed.customer_events
ADD COLUMNS (
    user_agent STRING,
    ip_address STRING
);

# PySpark để add column
spark.sql("""
    ALTER TABLE glue_catalog.lakehouse_processed.customer_events
    ADD COLUMNS (
        user_agent STRING COMMENT 'User agent string',
        ip_address STRING COMMENT 'Client IP address'
    )
""")

-- Rename column
ALTER TABLE lakehouse_processed.customer_events
RENAME COLUMN event_data TO event_payload;

-- Drop column
ALTER TABLE lakehouse_processed.customer_events
DROP COLUMN user_agent;

-- Widen column type
ALTER TABLE lakehouse_processed.customer_events
ALTER COLUMN event_id TYPE BIGINT;

-- Reorder columns
ALTER TABLE lakehouse_processed.customer_events
CHANGE COLUMN event_timestamp event_timestamp TIMESTAMP AFTER customer_id;

# test_schema_evolution.py
def test_schema_change(database, table, new_column):
    """
    Test schema evolution trong dev environment
    """
    spark = SparkSession.builder.getOrCreate()

    # Add column
    spark.sql(f"""
        ALTER TABLE {database}.{table}
        ADD COLUMNS ({new_column})
    """)

    # Test read with old and new schema
    df = spark.table(f"{database}.{table}")
    print(f"Schema after change: {df.schema}")

    # Test query
    result = spark.sql(f"""
        SELECT COUNT(*), COUNT({new_column.split()[0]})
        FROM {database}.{table}
    """)
    result.show()

# schema_change_log.py
import boto3
import json
from datetime import datetime

def log_schema_change(database, table, change_type, details):
    """
    Log schema changes to S3 for audit trail
    """
    s3_client = boto3.client('s3')

    log_entry = {
        'timestamp': datetime.now().isoformat(),
        'database': database,
        'table': table,
        'change_type': change_type,
        'details': details
    }

    log_key = f"schema-logs/{database}/{table}/{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"

    s3_client.put_object(
        Bucket='my-lakehouse-bucket',
        Key=log_key,
        Body=json.dumps(log_entry, indent=2)
    )

    print(f"Schema change logged to {log_key}")

# Usage
log_schema_change(
    database='lakehouse_processed',
    table='customer_events',
    change_type='ADD_COLUMN',
    details={'column_name': 'user_agent', 'column_type': 'STRING'}
)

# handle_schema_evolution.py
from pyspark.sql.functions import col, when, lit

def read_with_schema_evolution(database, table, default_values=None):
    """
    Read table với handling cho schema evolution
    """
    spark = SparkSession.builder.getOrCreate()
    df = spark.table(f"{database}.{table}")

    # Apply default values cho new columns nếu cần
    if default_values:
        for column, default_value in default_values.items():
            if column in df.columns:
                df = df.withColumn(
                    column,
                    when(col(column).isNull(), lit(default_value)).otherwise(col(column))
                )

    return df

# Usage
df = read_with_schema_evolution(
    database='lakehouse_processed',
    table='customer_events',
    default_values={'user_agent': 'unknown', 'ip_address': '0.0.0.0'}
)

-- Athena SQL để query data tại specific timestamp
SELECT *
FROM lakehouse_processed.customer_events
FOR SYSTEM_TIME AS OF TIMESTAMP '2024-01-15 10:00:00'
WHERE customer_id = 'CUST123';

-- Query specific snapshot
SELECT *
FROM lakehouse_processed.customer_events
FOR SYSTEM_VERSION AS OF 1234567890
LIMIT 100;

-- View all snapshots
SELECT
    snapshot_id,
    committed_at,
    operation,
    summary
FROM lakehouse_processed.customer_events.snapshots
ORDER BY committed_at DESC;

-- Athena SQL để rollback table to previous snapshot
CALL system.rollback_to_snapshot(
    table => 'lakehouse_processed.customer_events',
    snapshot_id => 1234567890
);

# PySpark để rollback
spark.sql("""
    CALL glue_catalog.system.rollback_to_snapshot(
        'lakehouse_processed.customer_events',
        1234567890
    )
""")

# snapshot_retention_policy.py
from datetime import datetime, timedelta

def apply_retention_policy(database, table, retention_days=7, milestone_day=1):
    """
    Apply snapshot retention policy
    - Keep all snapshots from last N days
    - Keep monthly milestone snapshots
    - Expire others
    """
    spark = SparkSession.builder.getOrCreate()

    # Get all snapshots
    snapshots_df = spark.sql(f"""
        SELECT
            snapshot_id,
            committed_at,
            EXTRACT(DAY FROM committed_at) as day_of_month
        FROM {database}.{table}.snapshots
    """)

    cutoff_date = datetime.now() - timedelta(days=retention_days)

    # Identify snapshots to keep
    snapshots_to_keep = []

    for row in snapshots_df.collect():
        snapshot_date = row['committed_at']

        # Keep recent snapshots
        if snapshot_date >= cutoff_date:
            snapshots_to_keep.append(row['snapshot_id'])
        # Keep milestone snapshots (first day of month)
        elif row['day_of_month'] == milestone_day:
            snapshots_to_keep.append(row['snapshot_id'])

    # Expire old snapshots (keeping identified ones)
    if snapshots_to_keep:
        # Iceberg will keep snapshots in the list
        oldest_to_expire = cutoff_date.strftime('%Y-%m-%d %H:%M:%S')

        spark.sql(f"""
            CALL glue_catalog.system.expire_snapshots(
                '{database}.{table}',
                TIMESTAMP '{oldest_to_expire}',
                {len(snapshots_to_keep)}
            )
        """)

        print(f"Applied retention policy. Kept {len(snapshots_to_keep)} snapshots.")

-- Athena SQL để change partition spec
ALTER TABLE lakehouse_processed.customer_events
SET PARTITION SPEC (days(event_date), bucket(10, customer_id));

# PySpark để change partition spec
spark.sql("""
    ALTER TABLE glue_catalog.lakehouse_processed.customer_events
    REPLACE PARTITION FIELD days(event_date) WITH days(event_date)
    REPLACE PARTITION FIELD customer_id WITH bucket(10, customer_id)
""")

-- User query (không cần partition predicate)
SELECT *
FROM lakehouse_processed.customer_events
WHERE event_date = DATE '2024-01-15'
  AND customer_id = 'CUST123';

-- Iceberg automatically prunes partitions based on event_date và customer_id

-- Query để analyze common filters
SELECT
    query_id,
    query_text,
    data_scanned_in_bytes / 1024 / 1024 / 1024 as data_scanned_gb,
    execution_time_millis / 1000 as execution_time_sec
FROM system.query_history
WHERE database_name = 'lakehouse_processed'
  AND table_name = 'customer_events'
ORDER BY execution_time_millis DESC
LIMIT 100;

# test_partition_evolution.py
def test_partition_change(database, table, new_partition_spec):
    """
    Test partition evolution impact
    """
    spark = SparkSession.builder.getOrCreate()

    # Measure query performance before change
    before_start = time.time()
    spark.sql(f"""
        SELECT COUNT(*)
        FROM {database}.{table}
        WHERE event_date >= DATE '2024-01-01'
    """).collect()
    before_time = time.time() - before_start

    # Apply partition change
    spark.sql(f"""
        ALTER TABLE {database}.{table}
        SET PARTITION SPEC {new_partition_spec}
    """)

    # Measure query performance after change
    after_start = time.time()
    spark.sql(f"""
        SELECT COUNT(*)
        FROM {database}.{table}
        WHERE event_date >= DATE '2024-01-01'
    """).collect()
    after_time = time.time() - after_start

    print(f"Query time before: {before_time:.2f}s")
    print(f"Query time after: {after_time:.2f}s")
    print(f"Performance change: {((after_time - before_time) / before_time * 100):.2f}%")

PARTITIONED BY (region)

PARTITIONED BY (bucket(10, customer_id))

PARTITIONED BY (truncate(5, user_id))

PARTITIONED BY (year(event_timestamp), month(event_timestamp))
-- hoặc
PARTITIONED BY (days(event_date))

CREATE TABLE lakehouse_processed.events (
    event_id STRING,
    event_timestamp TIMESTAMP,
    event_data MAP<STRING, STRING>
)
PARTITIONED BY (days(event_timestamp))
LOCATION 's3://my-lakehouse-bucket/processed/events/'
TBLPROPERTIES ('table_type' = 'ICEBERG');

CREATE TABLE lakehouse_processed.high_volume_events (
    event_id STRING,
    event_timestamp TIMESTAMP,
    event_data MAP<STRING, STRING>
)
PARTITIONED BY (hours(event_timestamp))
LOCATION 's3://my-lakehouse-bucket/processed/high_volume_events/'
TBLPROPERTIES ('table_type' = 'ICEBERG');

CREATE TABLE lakehouse_curated.monthly_aggregates (
    metric_name STRING,
    metric_value DOUBLE,
    aggregation_date DATE
)
PARTITIONED BY (months(aggregation_date))
LOCATION 's3://my-lakehouse-bucket/curated/monthly_aggregates/'
TBLPROPERTIES ('table_type' = 'ICEBERG');

CREATE TABLE lakehouse_processed.customer_transactions (
    transaction_id STRING,
    customer_id STRING,
    transaction_date DATE,
    region STRING,
    amount DOUBLE
)
PARTITIONED BY (days(transaction_date), region)
LOCATION 's3://my-lakehouse-bucket/processed/customer_transactions/'
TBLPROPERTIES ('table_type' = 'ICEBERG');

CREATE TABLE lakehouse_processed.user_activities (
    user_id STRING,
    activity_type STRING,
    activity_timestamp TIMESTAMP,
    activity_data MAP<STRING, STRING>
)
PARTITIONED BY (days(activity_timestamp), bucket(100, user_id))
LOCATION 's3://my-lakehouse-bucket/processed/user_activities/'
TBLPROPERTIES ('table_type' = 'ICEBERG');

-- BAD: Partitioning by hour cho low-volume data
CREATE TABLE lakehouse_processed.low_volume_events (
    event_id STRING,
    event_timestamp TIMESTAMP
)
PARTITIONED BY (hours(event_timestamp))  -- Tạo 24 partitions/day với very little data each

-- GOOD: Daily partitioning cho low-volume data
CREATE TABLE lakehouse_processed.low_volume_events (
    event_id STRING,
    event_timestamp TIMESTAMP
)
PARTITIONED BY (days(event_timestamp))  -- Reasonable partition size

-- BAD: Partitioning by customer_id (millions of unique values)
CREATE TABLE lakehouse_processed.customer_events (
    event_id STRING,
    customer_id STRING,
    event_timestamp TIMESTAMP
)
PARTITIONED BY (customer_id)  -- Millions of partitions!

-- GOOD: Bucketing customer_id
CREATE TABLE lakehouse_processed.customer_events (
    event_id STRING,
    customer_id STRING,
    event_timestamp TIMESTAMP
)
PARTITIONED BY (days(event_timestamp), bucket(100, customer_id))

s3://bucket/table/year=2024/month=01/day=15/file1.parquet
s3://bucket/table/year=2024/month=01/day=15/file2.parquet

# Iceberg automatically prunes partitions using metadata
# No need for S3 LIST operations
df = spark.sql("""
    SELECT *
    FROM lakehouse_processed.customer_events
    WHERE event_date = DATE '2024-01-15'
""")
# Iceberg reads manifest files to identify relevant data files
# Much faster than listing S3 prefixes

# partition_migration.py
def migrate_partition_scheme(database, table, new_partition_spec):
    """
    Safely migrate to new partition scheme
    """
    spark = SparkSession.builder.getOrCreate()

    # Step 1: Create new table với new partition spec
    temp_table = f"{table}_new"
    spark.sql(f"""
        CREATE TABLE {database}.{temp_table}
        USING iceberg
        PARTITIONED BY {new_partition_spec}
        AS SELECT * FROM {database}.{table}
    """)

    # Step 2: Validate data
    old_count = spark.sql(f"SELECT COUNT(*) FROM {database}.{table}").collect()[0][0]
    new_count = spark.sql(f"SELECT COUNT(*) FROM {database}.{temp_table}").collect()[0][0]

    if old_count != new_count:
        raise Exception(f"Data validation failed: {old_count} != {new_count}")

    # Step 3: Swap tables (requires manual intervention)
    print(f"Validation successful. Ready to swap {table} with {temp_table}")
    print("Manual steps:")
    print(f"1. DROP TABLE {database}.{table}")
    print(f"2. ALTER TABLE {database}.{temp_table} RENAME TO {table}")

# file_compaction.py
def compact_small_files(database, table, target_file_size_mb=512):
    """
    Compact small files to improve query performance
    """
    spark = SparkSession.builder.getOrCreate()

    # Analyze current file sizes
    files_df = spark.sql(f"""
        SELECT
            COUNT(*) as file_count,
            AVG(file_size_in_bytes) / 1024 / 1024 as avg_file_size_mb,
            MIN(file_size_in_bytes) / 1024 / 1024 as min_file_size_mb,
            MAX(file_size_in_bytes) / 1024 / 1024 as max_file_size_mb
        FROM {database}.{table}.files
    """)

    stats = files_df.collect()[0]
    print(f"Current stats: {stats}")

    # Run compaction if average file size is too small
    if stats['avg_file_size_mb'] < target_file_size_mb / 2:
        print(f"Running compaction with target size {target_file_size_mb}MB...")

        spark.sql(f"""
            CALL glue_catalog.system.rewrite_data_files(
                table => '{database}.{table}',
                options => map(
                    'target-file-size-bytes', '{target_file_size_mb * 1024 * 1024}',
                    'min-file-size-bytes', '10485760'  -- 10MB minimum
                )
            )
        """)

        print("Compaction completed")
    else:
        print("No compaction needed")

# Schedule compaction job
if __name__ == "__main__":
    compact_small_files('lakehouse_processed', 'customer_events', target_file_size_mb=512)

# glue_compaction_workflow.tf
resource "aws_glue_workflow" "compaction" {
  name = "${var.project_name}-compaction-workflow"

  description = "Automated file compaction workflow"
}

resource "aws_glue_trigger" "compaction_schedule" {
  name          = "${var.project_name}-compaction-trigger"
  type          = "SCHEDULED"
  schedule      = "cron(0 2 * * ? *)"  # Run daily at 2 AM
  workflow_name = aws_glue_workflow.compaction.name

  actions {
    job_name = aws_glue_job.compaction.name
  }
}

resource "aws_glue_job" "compaction" {
  name     = "${var.project_name}-compaction-job"
  role_arn = aws_iam_role.glue_job_role.arn

  command {
    name            = "glueetl"
    script_location = "s3://${aws_s3_bucket.lakehouse.id}/scripts/file_compaction.py"
    python_version  = "3"
  }

  default_arguments = {
    "--database_name"     = "lakehouse_processed"
    "--table_name"        = "customer_events"
    "--target_file_size"  = "536870912"  # 512MB
  }

  glue_version = "4.0"
  worker_type  = "G.1X"
  number_of_workers = 10
}

-- Use EXPLAIN để verify partition pruning
EXPLAIN
SELECT *
FROM lakehouse_processed.customer_events
WHERE event_date = DATE '2024-01-15'
  AND region = 'US';

-- Output should show partition filters applied

# monitor_partition_pruning.py
def analyze_query_efficiency(database, table, query):
    """
    Analyze query efficiency và partition pruning
    """
    athena_client = boto3.client('athena')

    # Execute query
    response = athena_client.start_query_execution(
        QueryString=query,
        QueryExecutionContext={'Database': database},
        ResultConfiguration={
            'OutputLocation': 's3://my-lakehouse-bucket/athena-results/'
        }
    )

    query_execution_id = response['QueryExecutionId']

    # Wait for completion
    while True:
        status_response = athena_client.get_query_execution(
            QueryExecutionId=query_execution_id
        )
        status = status_response['QueryExecution']['Status']['State']

        if status in ['SUCCEEDED', 'FAILED', 'CANCELLED']:
            break

        time.sleep(2)

    # Get statistics
    if status == 'SUCCEEDED':
        stats = status_response['QueryExecution']['Statistics']

        data_scanned_gb = stats['DataScannedInBytes'] / 1024 / 1024 / 1024
        execution_time_sec = stats['EngineExecutionTimeInMillis'] / 1000

        # Get total table size for comparison
        total_size_query = f"SELECT SUM(file_size_in_bytes) FROM {database}.{table}.files"
        total_size_response = athena_client.start_query_execution(
            QueryString=total_size_query,
            QueryExecutionContext={'Database': database},
            ResultConfiguration={
                'OutputLocation': 's3://my-lakehouse-bucket/athena-results/'
            }
        )

        # Calculate pruning efficiency
        print(f"Data scanned: {data_scanned_gb:.2f} GB")
        print(f"Execution time: {execution_time_sec:.2f} seconds")
        print(f"Query cost: ${data_scanned_gb * 0.005:.4f}")  # $5 per TB scanned

        return {
            'data_scanned_gb': data_scanned_gb,
            'execution_time_sec': execution_time_sec,
            'cost_usd': data_scanned_gb * 0.005
        }

# Usage
query = """
SELECT COUNT(*)
FROM lakehouse_processed.customer_events
WHERE event_date BETWEEN DATE '2024-01-01' AND DATE '2024-01-31'
  AND region = 'US'
"""

stats = analyze_query_efficiency('lakehouse_processed', 'customer_events', query)

# s3_lifecycle.tf
resource "aws_s3_bucket_lifecycle_configuration" "lakehouse_optimization" {
  bucket = aws_s3_bucket.lakehouse.id

  rule {
    id     = "raw-zone-lifecycle"
    status = "Enabled"

    filter {
      prefix = "raw/"
    }

    # Transition to IA after 30 days
    transition {
      days          = 30
      storage_class = "STANDARD_IA"
    }

    # Transition to Glacier after 90 days
    transition {
      days          = 90
      storage_class = "GLACIER_IR"
    }

    # Transition to Deep Archive after 365 days
    transition {
      days          = 365
      storage_class = "DEEP_ARCHIVE"
    }
  }

  rule {
    id     = "processed-zone-lifecycle"
    status = "Enabled"

    filter {
      prefix = "processed/"
    }

    # Keep processed data in Standard for 90 days
    transition {
      days          = 90
      storage_class = "STANDARD_IA"
    }

    # Archive old processed data
    transition {
      days          = 180
      storage_class = "GLACIER_IR"
    }
  }

  rule {
    id     = "curated-zone-lifecycle"
    status = "Enabled"

    filter {
      prefix = "curated/"
    }

    # Curated data stays in Standard longer
    transition {
      days          = 180
      storage_class = "STANDARD_IA"
    }
  }

  rule {
    id     = "cleanup-incomplete-uploads"
    status = "Enabled"

    abort_incomplete_multipart_upload {
      days_after_initiation = 7
    }
  }

  rule {
    id     = "cleanup-old-versions"
    status = "Enabled"

    noncurrent_version_transition {
      noncurrent_days = 30
      storage_class   = "STANDARD_IA"
    }

    noncurrent_version_expiration {
      noncurrent_days = 90
    }
  }
}

resource "aws_s3_bucket_intelligent_tiering_configuration" "lakehouse" {
  bucket = aws_s3_bucket.lakehouse.id
  name   = "EntireBucket"

  tiering {
    access_tier = "ARCHIVE_ACCESS"
    days        = 90
  }

  tiering {
    access_tier = "DEEP_ARCHIVE_ACCESS"
    days        = 180
  }
}

# write_with_compression.py
# Configure compression trong Spark
spark.conf.set("spark.sql.parquet.compression.codec", "snappy")

# Write data với compression
df.write \
    .format("iceberg") \
    .mode("append") \
    .option("write.parquet.compression-codec", "snappy") \
    .save("glue_catalog.lakehouse_processed.customer_events")

-- BAD: Full table scan
SELECT COUNT(*)
FROM lakehouse_processed.customer_events;
-- Scans: 1 TB, Cost: $5.00

-- GOOD: Partition pruning
SELECT COUNT(*)
FROM lakehouse_processed.customer_events
WHERE event_date = DATE '2024-01-15';
-- Scans: 10 GB, Cost: $0.05

-- BAD: Select all columns
SELECT *
FROM lakehouse_processed.customer_events
WHERE event_date = DATE '2024-01-15';
-- Scans: 10 GB

-- GOOD: Select specific columns
SELECT event_id, customer_id, event_type
FROM lakehouse_processed.customer_events
WHERE event_date = DATE '2024-01-15';
-- Scans: 2 GB (80% reduction)

# athena_query_cache.py
import boto3
import hashlib
import json

class AthenaQueryCache:
    def __init__(self, database, output_location):
        self.athena_client = boto3.client('athena')
        self.s3_client = boto3.client('s3')
        self.database = database
        self.output_location = output_location
        self.cache = {}

    def execute_with_cache(self, query, cache_ttl_hours=24):
        """
        Execute query với caching
        """
        # Generate cache key
        cache_key = hashlib.md5(query.encode()).hexdigest()

        # Check cache
        if cache_key in self.cache:
            cached_result = self.cache[cache_key]
            if self._is_cache_valid(cached_result, cache_ttl_hours):
                print("Returning cached result")
                return cached_result['data']

        # Execute query
        print("Executing new query")
        response = self.athena_client.start_query_execution(
            QueryString=query,
            QueryExecutionContext={'Database': self.database},
            ResultConfiguration={'OutputLocation': self.output_location}
        )

        query_execution_id = response['QueryExecutionId']

        # Wait and get results
        results = self._wait_and_get_results(query_execution_id)

        # Cache results
        self.cache[cache_key] = {
            'data': results,
            'timestamp': time.time()
        }

        return results

    def _is_cache_valid(self, cached_result, ttl_hours):
        """
        Check if cached result is still valid
        """
        age_hours = (time.time() - cached_result['timestamp']) / 3600
        return age_hours < ttl_hours

# athena_workgroup.tf
resource "aws_athena_workgroup" "lakehouse" {
  name = "${var.project_name}-workgroup"

  configuration {
    enforce_workgroup_configuration    = true
    publish_cloudwatch_metrics_enabled = true

    result_configuration {
      output_location = "s3://${aws_s3_bucket.lakehouse.id}/athena-results/"

      encryption_configuration {
        encryption_option = "SSE_S3"
      }
    }

    # Set data scanned limit
    bytes_scanned_cutoff_per_query = 1099511627776  # 1 TB limit
  }

  tags = {
    Name        = "Lakehouse Workgroup"
    Environment = var.environment
  }
}

# Workgroup for development (lower limits)
resource "aws_athena_workgroup" "dev" {
  name = "${var.project_name}-dev-workgroup"

  configuration {
    enforce_workgroup_configuration    = true
    publish_cloudwatch_metrics_enabled = true

    result_configuration {
      output_location = "s3://${aws_s3_bucket.lakehouse.id}/athena-results/dev/"
    }

    # Lower limit for dev
    bytes_scanned_cutoff_per_query = 107374182400  # 100 GB limit
  }
}

# optimize_glue_job.py
def calculate_optimal_workers(input_size_gb, processing_time_target_min):
    """
    Calculate optimal number of workers cho Glue job

    Rule of thumb: 1 G.1X worker can process ~10 GB/hour
    """
    # Calculate required processing rate
    required_rate_gb_per_hour = (input_size_gb / processing_time_target_min) * 60

    # Calculate workers needed
    workers_needed = max(2, int(required_rate_gb_per_hour / 10))

    # Add 20% buffer
    workers_with_buffer = int(workers_needed * 1.2)

    return {
        'recommended_workers': workers_with_buffer,
        'worker_type': 'G.1X',
        'estimated_cost': workers_with_buffer * 0.44 * (processing_time_target_min / 60)
    }

# Example
result = calculate_optimal_workers(input_size_gb=500, processing_time_target_min=30)
print(f"Recommended configuration: {result}")

# emr_autoscaling.tf
resource "aws_emr_cluster" "lakehouse" {
  name          = "${var.project_name}-emr-cluster"
  release_label = "emr-6.10.0"
  applications  = ["Spark", "Hadoop"]

  ec2_attributes {
    subnet_id                         = var.subnet_id
    emr_managed_master_security_group = aws_security_group.emr_master.id
    emr_managed_slave_security_group  = aws_security_group.emr_slave.id
    instance_profile                  = aws_iam_instance_profile.emr_profile.arn
  }

  master_instance_group {
    instance_type = "m5.xlarge"
  }

  core_instance_group {
    instance_type  = "m5.xlarge"
    instance_count = 2

    autoscaling_policy = jsonencode({
      Constraints = {
        MinCapacity = 2
        MaxCapacity = 10
      }
      Rules = [
        {
          Name = "ScaleOutOnYARNMemory"
          Description = "Scale out when YARN memory is high"
          Action = {
            SimpleScalingPolicyConfiguration = {
              AdjustmentType = "CHANGE_IN_CAPACITY"
              ScalingAdjustment = 2
              CoolDown = 300
            }
          }
          Trigger = {
            CloudWatchAlarmDefinition = {
              ComparisonOperator = "LESS_THAN"
              EvaluationPeriods = 1
              MetricName = "YARNMemoryAvailablePercentage"
              Namespace = "AWS/ElasticMapReduce"
              Period = 300
              Statistic = "AVERAGE"
              Threshold = 15.0
              Unit = "PERCENT"
            }
          }
        },
        {
          Name = "ScaleInOnYARNMemory"
          Description = "Scale in when YARN memory is low"
          Action = {
            SimpleScalingPolicyConfiguration = {
              AdjustmentType = "CHANGE_IN_CAPACITY"
              ScalingAdjustment = -1
              CoolDown = 300
            }
          }
          Trigger = {
            CloudWatchAlarmDefinition = {
              ComparisonOperator = "GREATER_THAN"
              EvaluationPeriods = 1
              MetricName = "YARNMemoryAvailablePercentage"
              Namespace = "AWS/ElasticMapReduce"
              Period = 300
              Statistic = "AVERAGE"
              Threshold = 75.0
              Unit = "PERCENT"
            }
          }
        }
      ]
    })
  }

  service_role = aws_iam_role.emr_service_role.arn
}

# cost_allocation_tags.tf
resource "aws_s3_bucket_tagging" "lakehouse" {
  bucket = aws_s3_bucket.lakehouse.id

  tagging_rule {
    tag {
      key   = "CostCenter"
      value = "DataEngineering"
    }
    tag {
      key   = "Project"
      value = var.project_name
    }
    tag {
      key   = "Environment"
      value = var.environment
    }
    tag {
      key   = "DataZone"
      value = "Lakehouse"
    }
  }
}

# cost_monitoring.py
import boto3
from datetime import datetime, timedelta

def monitor_athena_costs(lookback_days=7, threshold_usd=100):
    """
    Monitor Athena query costs và alert on anomalies
    """
    ce_client = boto3.client('ce')  # Cost Explorer
    cloudwatch = boto3.client('cloudwatch')

    # Get cost data
    end_date = datetime.now().date()
    start_date = end_date - timedelta(days=lookback_days)

    response = ce_client.get_cost_and_usage(
        TimePeriod={
            'Start': start_date.strftime('%Y-%m-%d'),
            'End': end_date.strftime('%Y-%m-%d')
        },
        Granularity='DAILY',
        Metrics=['UnblendedCost'],
        Filter={
            'Dimensions': {
                'Key': 'SERVICE',
                'Values': ['Amazon Athena']
            }
        }
    )

    # Analyze costs
    daily_costs = []
    for result in response['ResultsByTime']:
        cost = float(result['Total']['UnblendedCost']['Amount'])
        daily_costs.append(cost)

        # Publish to CloudWatch
        cloudwatch.put_metric_data(
            Namespace='Lakehouse/Costs',
            MetricData=[
                {
                    'MetricName': 'AthenaDailyCost',
                    'Value': cost,
                    'Unit': 'None',
                    'Timestamp': datetime.strptime(result['TimePeriod']['Start'], '%Y-%m-%d')
                }
            ]
        )

    # Check for anomalies
    avg_cost = sum(daily_costs) / len(daily_costs)
    max_cost = max(daily_costs)

    if max_cost > threshold_usd:
        print(f"⚠️ ALERT: Daily Athena cost exceeded threshold: ${max_cost:.2f}")
        # Send SNS notification
        sns_client = boto3.client('sns')
        sns_client.publish(
            TopicArn='arn:aws:sns:us-east-1:123456789012:lakehouse-cost-alerts',
            Subject='Athena Cost Alert',
            Message=f'Daily Athena cost exceeded ${threshold_usd}: ${max_cost:.2f}'
        )

    return {
        'average_daily_cost': avg_cost,
        'max_daily_cost': max_cost,
        'total_cost': sum(daily_costs)
    }