# S3에서는 "디렉토리"가 실제로 존재하지 않음
# 모든 것이 Key-Value 쌍
s3://bucket/data/events/year=2024/month=01/day=15/part-00000.parquet
# 위는 실제로 하나의 긴 Key일 뿐

import boto3

s3 = boto3.client('s3')

# yyyy/mm/dd 구조에서 특정 날짜 데이터 찾기
# 1. year=2024 리스트 -> API 호출 1회
# 2. month=01 리스트 -> API 호출 1회  
# 3. day=15 리스트 -> API 호출 1회
# 총 3번의 API 호출 + 네트워크 레이턴시

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

# 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")
# 결과: Partition discovery: 5.43s (깊은 구조)
# vs
# 결과: Partition discovery: 0.87s (단순 구조)

# Spark에서 단순 파티션 구조 생성
df.write \
    .partitionBy("date") \
    .parquet("s3://bucket/data/events/")

# date 컬럼을 yyyy-mm-dd 형식으로 준비
from pyspark.sql.functions import date_format

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

# Spark에서 파일 크기 제어
spark.conf.set("spark.sql.files.maxRecordsPerFile", 1000000)
spark.conf.set("spark.sql.files.maxPartitionBytes", 134217728)  # 128MB

# repartition으로 파일 수 조정
df.repartition(100) \
    .write \
    .partitionBy("date") \
    .parquet("s3://bucket/data/events/")

# Small files를 큰 파일로 병합
from pyspark.sql import SparkSession

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

# 기존 작은 파일들 읽기
df = spark.read.parquet("s3://bucket/data/events/date=2024-01-15/")

# 적절한 파티션 수로 재작성
num_partitions = max(1, int(df.count() / 1000000))  # 파티션당 100만 레코드

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

# hash 기반 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 형식 (권장)
df = df.withColumn("date", date_format("event_time", "yyyy-MM-dd"))

# yyyymmdd 형식 (더 간결)
df = df.withColumn("date", date_format("event_time", "yyyyMMdd"))

# 파티셔닝
df.write \
    .partitionBy("date") \
    .parquet("s3://bucket/data/events/")

# yyyy/mm/dd 구조 (안티패턴)
{
  "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 구조 (최적화)
{
  "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
}

# 기존 파티션 구조 분석
import boto3
from collections import defaultdict

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

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

# 파티션별 파일 수와 크기 집계
stats = defaultdict(lambda: {"count": 0, "size": 0})

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

# 통계 출력
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()

# 기존 데이터 읽기
source_path = "s3://bucket/data/events-old/year=*/month=*/day=*/"
df = spark.read.parquet(source_path)

# 새로운 date 컬럼 생성
df = df.withColumn("date", date_format(col("event_time"), "yyyy-MM-dd"))

# Shard 추가 (선택사항)
df = df.withColumn("shard", abs(hash(col("user_id"))) % 10)

# 파일 크기 최적화
spark.conf.set("spark.sql.files.maxPartitionBytes", 134217728)  # 128MB

# 적절한 파티션 수 계산
total_size_gb = df.count() * 500 / 1024 / 1024 / 1024  # 레코드당 ~500 bytes
num_partitions = int(total_size_gb * 8)  # 128MB 파일 기준

# 새로운 구조로 저장
df.repartition(num_partitions, "date") \
    .write \
    .mode("overwrite") \
    .partitionBy("date") \
    .parquet("s3://bucket/data/events-new/")

# 날짜별로 점진적 마이그레이션
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}...")
    
    # 특정 날짜 데이터 읽기
    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))
        
        # 파일 크기 최적화
        num_files = max(1, int(df.count() / 1000000))  # 파일당 100만 레코드
        
        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)

# 마이그레이션 검증 스크립트
def validate_migration(old_path, new_path, date):
    # 레코드 수 비교
    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()
    
    # 체크섬 비교 (샘플링)
    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]
    
    # 결과
    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

# 전체 기간 검증
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)

# CloudWatch 메트릭 수집
import boto3
from datetime import datetime, timedelta

cloudwatch = boto3.client('cloudwatch')

# Athena 쿼리 실행 시간 모니터링
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']
)

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

# 오래된 파티션을 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)
    
    # 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)