{
  "Rules": [
    {
      "ID": "DataLakeLifecycle",
      "Status": "Enabled",
      "Filter": {"Prefix": "raw/"},
      "Transitions": [
        {"Days": 0, "StorageClass": "STANDARD"},
        {"Days": 90, "StorageClass": "STANDARD_IA"},
        {"Days": 180, "StorageClass": "GLACIER"},
        {"Days": 365, "StorageClass": "DEEP_ARCHIVE"}
      ],
      "Expiration": {"Days": 2555}
    },
    {
      "ID": "CleanupIncompleteUploads",
      "Status": "Enabled",
      "Filter": {},
      "AbortIncompleteMultipartUpload": {"DaysAfterInitiation": 7}
    }
  ]
}

{
  "Rules": [
    {
      "ApplyServerSideEncryptionByDefault": {
        "SSEAlgorithm": "aws:kms",
        "KMSMasterKeyID": "arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012"
      },
      "BucketKeyEnabled": true
    }
  ]
}

import boto3
from boto3.s3.transfer import TransferConfig

s3 = boto3.client('s3')

config = TransferConfig(
    multipart_threshold=1024 * 1024 * 100,
    max_concurrency=10,
    multipart_chunksize=1024 * 1024 * 100,
    use_threads=True
)

s3.upload_file(
    'large_dataset.parquet',
    'data-lake-bucket',
    'raw/transactions/2024/01/15/file.parquet',
    Config=config
)

import boto3
import json
from datetime import datetime, timedelta
from botocore.exceptions import ClientError

class S3DataManager:
    """Manages S3 data lake operations for data engineering."""

    def __init__(self, region='us-east-1'):
        self.region = region
        self.s3 = boto3.client('s3', region_name=region)

    def create_data_lake_bucket(self, bucket_name):
        """Create S3 bucket with versioning and encryption enabled."""
        try:
            if self.region == 'us-east-1':
                self.s3.create_bucket(Bucket=bucket_name)
            else:
                self.s3.create_bucket(
                    Bucket=bucket_name,
                    CreateBucketConfiguration={'LocationConstraint': self.region}
                )

            self.s3.put_bucket_versioning(
                Bucket=bucket_name,
                VersioningConfiguration={'Status': 'Enabled'}
            )

            self.s3.put_bucket_encryption(
                Bucket=bucket_name,
                ServerSideEncryptionConfiguration={
                    'Rules': [{
                        'ApplyServerSideEncryptionByDefault': {'SSEAlgorithm': 'aws:kms'},
                        'BucketKeyEnabled': True
                    }]
                }
            )

            self.s3.put_public_access_block(
                Bucket=bucket_name,
                PublicAccessBlockConfiguration={
                    'BlockPublicAcls': True,
                    'IgnorePublicAcls': True,
                    'BlockPublicPolicy': True,
                    'RestrictPublicBuckets': True
                }
            )

            self.s3.put_bucket_lifecycle_configuration(
                Bucket=bucket_name,
                LifecycleConfiguration={
                    'Rules': [
                        {
                            'ID': 'RawDataLifecycle',
                            'Status': 'Enabled',
                            'Filter': {'Prefix': 'raw/'},
                            'Transitions': [
                                {'Days': 90, 'StorageClass': 'STANDARD_IA'},
                                {'Days': 180, 'StorageClass': 'GLACIER'},
                                {'Days': 365, 'StorageClass': 'DEEP_ARCHIVE'}
                            ]
                        },
                        {
                            'ID': 'CleanupIncompleteUploads',
                            'Status': 'Enabled',
                            'Filter': {},
                            'AbortIncompleteMultipartUpload': {'DaysAfterInitiation': 7}
                        }
                    ]
                }
            )

            return bucket_name
        except ClientError as e:
            print(f"Error creating bucket: {e.response['Error']['Message']}")
            raise

    def upload_with_multipart(self, bucket, key, file_path, chunk_size=100*1024*1024):
        """Upload large files using multipart upload."""
        from boto3.s3.transfer import TransferConfig

        config = TransferConfig(
            multipart_threshold=chunk_size,
            max_concurrency=10,
            multipart_chunksize=chunk_size,
            use_threads=True
        )

        try:
            self.s3.upload_file(
                file_path, bucket, key, Config=config
            )
            print(f"Uploaded {file_path} to s3://{bucket}/{key}")
        except ClientError as e:
            print(f"Error uploading file: {e.response['Error']['Message']}")
            raise

    def list_partitioned_objects(self, bucket, prefix):
        """List objects in a partitioned structure."""
        paginator = self.s3.get_paginator('list_objects_v2')
        objects = []

        for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
            for obj in page.get('Contents', []):
                objects.append({
                    'key': obj['Key'],
                    'size': obj['Size'],
                    'last_modified': obj['LastModified']
                })

        return objects

    def compact_small_files(self, bucket, prefix, target_size_mb=256):
        """Identify and report small files for compaction."""
        objects = self.list_partitioned_objects(bucket, prefix)
        small_files = [obj for obj in objects if obj['size'] < target_size_mb * 1024 * 1024]

        print(f"Found {len(small_files)} small files (under {target_size_mb}MB)")
        for obj in small_files[:10]:
            print(f"  {obj['key']}: {obj['size'] / 1024 / 1024:.2f} MB")

        return small_files

if __name__ == '__main__':
    manager = S3DataManager(region='us-east-1')

    bucket = manager.create_data_lake_bucket('my-data-lake-prod-2024')
    print(f"Created bucket: {bucket}")

    manager.upload_with_multipart(
        bucket,
        'raw/transactions/2024/01/15/data.parquet',
        './large_dataset.parquet'
    )

    small_files = manager.compact_small_files(bucket, 'raw/transactions/')

def calculate_storage_cost(total_tb, storage_class='standard', months=1):
    """
    Calculate monthly storage cost.

    Standard: $0.023/GB/month
    IA: $0.0125/GB/month
    Glacier Instant: $0.004/GB/month
    Glacier Flexible: $0.0036/GB/month
    Deep Archive: $0.00099/GB/month
    """
    total_gb = total_tb * 1024
    cost_per_gb = {
        'standard': 0.023,
        'ia': 0.0125,
        'one_zone_ia': 0.01,
        'glacier_instant': 0.004,
        'glacier_flexible': 0.0036,
        'deep_archive': 0.00099
    }.get(storage_class, 0.023)

    return round(total_gb * cost_per_gb * months, 2)

# 10 TB in Standard: $235.52/month
print(f"Standard: ${calculate_storage_cost(10, 'standard')}")
# 10 TB in Deep Archive: $10.14/month
print(f"Deep Archive: ${calculate_storage_cost(10, 'deep_archive')}")

def calculate_lifecycle_savings(data_tb, hot_pct=0.2, warm_pct=0.3, cold_pct=0.5):
    """
    Calculate savings from lifecycle policies.

    Hot (Standard): 20% of data
    Warm (IA): 30% of data
    Cold (Glacier): 50% of data
    """
    total_gb = data_tb * 1024

    hot_gb = total_gb * hot_pct
    warm_gb = total_gb * warm_pct
    cold_gb = total_gb * cold_pct

    all_standard = total_gb * 0.023
    with_lifecycle = (hot_gb * 0.023 + warm_gb * 0.0125 + cold_gb * 0.0036)

    savings = all_standard - with_lifecycle
    savings_pct = (savings / all_standard) * 100

    return round(savings, 2), round(savings_pct, 1)

# 100 TB data lake
savings, pct = calculate_lifecycle_savings(100)
print(f"Monthly savings: ${savings} ({pct}%)")

def calculate_multipart_efficiency(file_size_gb, chunk_size_mb=100):
    """
    Calculate multipart upload efficiency.

    Parallel upload of chunks reduces total transfer time.
    """
    total_chunks = (file_size_gb * 1024) / chunk_size_mb
    parallelism = 10

    sequential_time = file_size_gb * 60  # rough estimate: 60s per GB
    parallel_time = sequential_time / parallelism

    return {
        'total_chunks': int(total_chunks),
        'sequential_time_min': round(sequential_time / 60, 1),
        'parallel_time_min': round(parallel_time / 60, 1),
        'efficiency_gain': f"{((sequential_time - parallel_time) / sequential_time) * 100:.0f}%"
    }