Amazon S3 — Complete Reference
for Data Engineers

# S3 Lifecycle Policy — BigData Datalake
{
  "Rules": [
    {
      "ID": "DataLakeLifecycle",
      "Status": "Enabled",
      "Filter": { "Prefix": "data/processed/" },
      "Transitions": [
        {
          "Days": 30,
          "StorageClass": "STANDARD_IA"    // After 30 days
        },
        {
          "Days": 90,
          "StorageClass": "GLACIER_IR"      // After 90 days
        },
        {
          "Days": 365,
          "StorageClass": "DEEP_ARCHIVE"    // After 1 year
        }
      ],
      "NoncurrentVersionTransitions": [
        { "NoncurrentDays": 7, "StorageClass": "STANDARD_IA" }
      ],
      "NoncurrentVersionExpiration": { "NoncurrentDays": 90 }
    }
  ]
}

# ═══ S3 Data Lake — Production Setup ═══

resource "aws_s3_bucket" "datalake_curated" {
  bucket = "my-company-curated-prod"

  tags = {
    Environment = "production"
    DataClass   = "curated"
    Team        = "data-platform"
  }
}

# ── Block all public access ──
resource "aws_s3_bucket_public_access_block" "curated" {
  bucket                  = aws_s3_bucket.datalake_curated.id
  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

# ── SSE-KMS Encryption ──
resource "aws_s3_bucket_server_side_encryption_configuration" "curated" {
  bucket = aws_s3_bucket.datalake_curated.id
  rule {
    apply_server_side_encryption_by_default {
      sse_algorithm     = "aws:kms"
      kms_master_key_id = aws_kms_key.datalake.arn
    }
    bucket_key_enabled = true  # Reduces KMS API costs by 99%
  }
}

# ── Versioning ──
resource "aws_s3_bucket_versioning" "curated" {
  bucket = aws_s3_bucket.datalake_curated.id
  versioning_configuration { status = "Enabled" }
}

# ── Lifecycle Policy ──
resource "aws_s3_bucket_lifecycle_configuration" "curated" {
  bucket = aws_s3_bucket.datalake_curated.id

  rule {
    id     = "tiering"
    status = "Enabled"

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

    noncurrent_version_expiration {
      noncurrent_days = 90
    }
  }
}

# ── CRR to DR Region ──
resource "aws_s3_bucket_replication_configuration" "curated" {
  bucket = aws_s3_bucket.datalake_curated.id
  role   = aws_iam_role.replication.arn

  rule {
    id     = "disaster-recovery"
    status = "Enabled"

    destination {
      bucket        = aws_s3_bucket.datalake_curated_dr.arn
      storage_class = "STANDARD_IA"

      replication_time {
        status = "Enabled"   # RTC: 15-min SLA
        time { minutes = 15 }
      }
    }
  }
}

import boto3
from boto3.s3.transfer import TransferConfig
import concurrent.futures

# ── Optimized config for PB-scale uploads ──
transfer_config = TransferConfig(
    multipart_threshold = 100 * 1024 * 1024,   # 100 MB
    multipart_chunksize = 128 * 1024 * 1024,   # 128 MB parts
    max_concurrency     = 20,                   # 20 threads/object
    use_threads         = True
)

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

def upload_to_datalake(local_path: str, s3_key: str):
    """Upload with optimised multipart + SSE-KMS + metadata."""
    s3.upload_file(
        Filename   = local_path,
        Bucket     = 'my-company-raw-prod',
        Key        = s3_key,
        Config     = transfer_config,
        ExtraArgs  = {
            'ServerSideEncryption': 'aws:kms',
            'SSEKMSKeyId'        : 'arn:aws:kms:...',
            'StorageClass'       : 'INTELLIGENT_TIERING',
            'ContentType'        : 'application/octet-stream',
            'Metadata'           : {
                'pipeline-version': 'v2.1',
                'source-system'   : 'kafka-prod'
            }
        }
    )

# ── Paginated LIST across billions of objects ──
def list_objects_paginated(bucket: str, prefix: str):
    paginator = s3.get_paginator('list_objects_v2')
    pages = paginator.paginate(Bucket=bucket, Prefix=prefix)
    for page in pages:
        for obj in page.get('Contents', []):
            yield obj['Key'], obj['Size']

# ── Byte-range GET (read only a Parquet row group) ──
def read_parquet_footer(bucket: str, key: str) -> bytes:
    response = s3.get_object(
        Bucket    = bucket,
        Key       = key,
        Range     = 'bytes=-8192'  # Last 8KB (Parquet footer)
    )
    return response['Body'].read()