# "Directories" don't actually exist in S3
# Everything is a Key-Value pair
s3://bucket/data/events/year=2024/month=01/day=15/part-00000.parquet
# This is actually just one long Key

import boto3

s3 = boto3.client('s3')

# Finding data for a specific date in yyyy/mm/dd structure
# 1. List year=2024 -> API call 1
# 2. List month=01 -> API call 1
# 3. List day=15 -> API call 1
# Total 3 API calls + network latency

response = s3.list_objects_v2(
    Bucket='my-bucket',
    Prefix='data/events/year=2024/month=01/day=15/'
)

# Measuring partition discovery time in Spark
import time

start = time.time()
df = spark.read.parquet("s3://bucket/data/events/year=2024/month=01/day=15/")
print(f"Partition discovery: {time.time() - start:.2f}s")
# Result: Partition discovery: 5.43s (deep structure)
# vs
# Result: Partition discovery: 0.87s (simple structure)

# Create simple partition structure in Spark
df.write \
    .partitionBy("date") \
    .parquet("s3://bucket/data/events/")

# Prepare date column in yyyy-mm-dd format
from pyspark.sql.functions import date_format

df = df.withColumn("date", date_format("timestamp", "yyyy-MM-dd"))

# Control file size in Spark
spark.conf.set("spark.sql.files.maxRecordsPerFile", 1000000)
spark.conf.set("spark.sql.files.maxPartitionBytes", 134217728)  # 128MB

# Adjust file count with repartition
df.repartition(100) \
    .write \
    .partitionBy("date") \
    .parquet("s3://bucket/data/events/")

# Merge small files into large files
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("S3 Compaction") \
    .getOrCreate()

# Read existing small files
df = spark.read.parquet("s3://bucket/data/events/date=2024-01-15/")

# Calculate appropriate partition count
num_partitions = max(1, int(df.count() / 1000000))  # 1M records per partition

# Rewrite with appropriate partition count
df.repartition(num_partitions) \
    .write \
    .mode("overwrite") \
    .parquet("s3://bucket/data/events-compacted/date=2024-01-15/")

# Create hash-based shard
from pyspark.sql.functions import hash, abs, col

df = df.withColumn("shard", abs(hash(col("user_id"))) % 10)

df.write \
    .partitionBy("date", "shard") \
    .parquet("s3://bucket/data/events/")

from pyspark.sql.functions import date_format

# yyyy-mm-dd format (recommended)
df = df.withColumn("date", date_format("event_time", "yyyy-MM-dd"))

# yyyymmdd format (more concise)
df = df.withColumn("date", date_format("event_time", "yyyyMMdd"))

# Partitioning
df.write \
    .partitionBy("date") \
    .parquet("s3://bucket/data/events/")

# yyyy/mm/dd structure (anti-pattern)
{
  "partition_discovery_ms": 5430,
  "s3_list_calls": 4,
  "s3_get_calls": 720,
  "network_latency_ms": 18200,
  "data_read_mb": 2880,
  "executor_time_s": 48.3
}

# yyyy-mm-dd structure (optimized)
{
  "partition_discovery_ms": 870,
  "s3_list_calls": 2,
  "s3_get_calls": 24,
  "network_latency_ms": 2400,
  "data_read_mb": 3072,
  "executor_time_s": 12.1
}

# Analyze existing partition structure
import boto3
from collections import defaultdict

s3 = boto3.client('s3')
paginator = s3.get_paginator('list_objects_v2')

bucket = 'my-bucket'
prefix = 'data/events/'

# Aggregate file count and size per partition
stats = defaultdict(lambda: {"count": 0, "size": 0})

for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
    for obj in page.get('Contents', []):
        # Extract partition (e.g., year=2024/month=01/day=15)
        parts = obj['Key'].split('/')
        partition = '/'.join(parts[:-1])
        
        stats[partition]["count"] += 1
        stats[partition]["size"] += obj['Size']

# Output statistics
for partition, data in sorted(stats.items()):
    avg_size_mb = data["size"] / data["count"] / 1024 / 1024
    print(f"{partition}: {data['count']} files, avg {avg_size_mb:.1f}MB")

from pyspark.sql import SparkSession
from pyspark.sql.functions import date_format, abs, hash, col

spark = SparkSession.builder \
    .appName("S3 Partition Migration") \
    .config("spark.sql.sources.partitionOverwriteMode", "dynamic") \
    .getOrCreate()

# Read existing data
source_path = "s3://bucket/data/events-old/year=*/month=*/day=*/"
df = spark.read.parquet(source_path)

# Create new date column
df = df.withColumn("date", date_format(col("event_time"), "yyyy-MM-dd"))

# Add shard (optional)
df = df.withColumn("shard", abs(hash(col("user_id"))) % 10)

# Optimize file size
spark.conf.set("spark.sql.files.maxPartitionBytes", 134217728)  # 128MB

# Calculate appropriate partition count
total_size_gb = df.count() * 500 / 1024 / 1024 / 1024  # ~500 bytes per record
num_partitions = int(total_size_gb * 8)  # Based on 128MB files

# Save with new structure
df.repartition(num_partitions, "date") \
    .write \
    .mode("overwrite") \
    .partitionBy("date") \
    .parquet("s3://bucket/data/events-new/")

# Incremental migration by date
from datetime import datetime, timedelta

start_date = datetime(2024, 1, 1)
end_date = datetime(2024, 12, 31)
current_date = start_date

while current_date <= end_date:
    year = current_date.year
    month = current_date.month
    day = current_date.day
    date_str = current_date.strftime("%Y-%m-%d")
    
    print(f"Processing {date_str}...")
    
    # Read data for specific date
    old_path = f"s3://bucket/data/events-old/year={year}/month={month:02d}/day={day:02d}/"
    
    try:
        df = spark.read.parquet(old_path)
        df = df.withColumn("date", lit(date_str))
        
        # Optimize file size
        num_files = max(1, int(df.count() / 1000000))  # 1M records per file
        
        df.repartition(num_files) \
            .write \
            .mode("overwrite") \
            .parquet(f"s3://bucket/data/events-new/date={date_str}/")
        
        print(f"✓ {date_str} completed")
    except Exception as e:
        print(f"✗ {date_str} failed: {e}")
    
    current_date += timedelta(days=1)

# Migration validation script
def validate_migration(old_path, new_path, date):
    # Compare record counts
    old_df = spark.read.parquet(f"{old_path}/year={date.year}/month={date.month:02d}/day={date.day:02d}/")
    new_df = spark.read.parquet(f"{new_path}/date={date.strftime('%Y-%m-%d')}/")
    
    old_count = old_df.count()
    new_count = new_df.count()
    
    # Compare checksums (sampling)
    old_checksum = old_df.sample(0.01).selectExpr("sum(hash(*))").collect()[0][0]
    new_checksum = new_df.sample(0.01).selectExpr("sum(hash(*))").collect()[0][0]
    
    # Result
    if old_count == new_count and old_checksum == new_checksum:
        print(f"✓ {date}: Valid ({old_count:,} records)")
        return True
    else:
        print(f"✗ {date}: Invalid (old: {old_count:,}, new: {new_count:,})")
        return False

# Validate entire period
from datetime import datetime, timedelta

start_date = datetime(2024, 1, 1)
end_date = datetime(2024, 1, 31)
current_date = start_date

while current_date <= end_date:
    validate_migration(
        "s3://bucket/data/events-old",
        "s3://bucket/data/events-new",
        current_date
    )
    current_date += timedelta(days=1)

# Collect CloudWatch metrics
import boto3
from datetime import datetime, timedelta

cloudwatch = boto3.client('cloudwatch')

# Monitor Athena query execution time
response = cloudwatch.get_metric_statistics(
    Namespace='AWS/Athena',
    MetricName='EngineExecutionTime',
    Dimensions=[
        {'Name': 'WorkGroup', 'Value': 'primary'}
    ],
    StartTime=datetime.now() - timedelta(days=7),
    EndTime=datetime.now(),
    Period=3600,
    Statistics=['Average', 'Maximum']
)

# Analyze results
for datapoint in response['Datapoints']:
    print(f"{datapoint['Timestamp']}: "
          f"Avg={datapoint['Average']:.2f}s, "
          f"Max={datapoint['Maximum']:.2f}s")

# Transition old partitions to Intelligent-Tiering
import boto3

s3 = boto3.client('s3')

def transition_old_partitions(bucket, prefix, days_old=90):
    cutoff_date = datetime.now() - timedelta(days=days_old)
    
    # Create lifecycle policy
    lifecycle_config = {
        'Rules': [
            {
                'Id': 'TransitionOldData',
                'Status': 'Enabled',
                'Prefix': prefix,
                'Transitions': [
                    {
                        'Days': days_old,
                        'StorageClass': 'INTELLIGENT_TIERING'
                    }
                ]
            }
        ]
    }
    
    s3.put_bucket_lifecycle_configuration(
        Bucket=bucket,
        LifecycleConfiguration=lifecycle_config
    )
    
    print(f"✓ Lifecycle policy applied: {days_old}+ days → INTELLIGENT_TIERING")

transition_old_partitions('my-bucket', 'data/events/', 90)