Working with Files at Scale

# ── THE RULES ────────────────────────────────────────────────────────────────
# 1. Timestamps: always ISO 8601, always UTC, always in the name
# 2. Use underscores not spaces (spaces cause shell escaping nightmares)
# 3. Include enough context to identify without opening the file
# 4. Lexicographic sort order = chronological sort order (ISO dates achieve this)
# 5. Include a unique identifier to prevent overwrite collisions

# ── LANDING ZONE: preserve origin context ────────────────────────────────────
# Pattern: {source}_{entity}_{start_ts}_{end_ts}_{batch_id}.{ext}
stripe_payments_20260317T000000Z_20260317T235959Z_f8a3b2c4.json
shopify_orders_20260317T060000Z_20260317T120000Z_9e1d7c3f.csv
freshcart_deliveries_20260317T000000Z_20260317T235959Z_2b4a8d6e.parquet

# What each component gives you:
# stripe       → which source system (filter by prefix)
# payments     → which entity/table
# 20260317T... → ISO 8601 UTC timestamps (sort = chronological order)
# f8a3b2c4     → unique batch ID (trace back to pipeline run logs)

# ── BRONZE LAYER: add pipeline metadata ──────────────────────────────────────
# Pattern: {entity}/year={YYYY}/month={MM}/day={DD}/{entity}_{ts}_{id}.parquet
# (Hive-style partitioning — covered in Part 03)

s3://freshcart-lake/bronze/payments/year=2026/month=03/day=17/
  payments_20260317T000000Z_f8a3b2c4.parquet
  payments_20260317T060000Z_9e1d7c3f.parquet

# ── SILVER AND GOLD: clean entity-oriented names ─────────────────────────────
s3://freshcart-lake/silver/orders/date=2026-03-17/part-00001.parquet
s3://freshcart-lake/gold/daily_revenue/date=2026-03-17/part-00001.parquet

# BAD: no timestamp — cannot determine when file was created
orders.csv

# BAD: ambiguous date format — is 03/17/26 March or day 3?
orders_03-17-26.csv

# BAD: spaces in filename — breaks shell commands without quoting
march 17 orders.csv

# BAD: no source or entity context
data_f8a3b2c4.parquet

# BAD: non-sortable timestamp — file 2_11_2026 sorts before 3_1_2026
orders_2_11_2026.csv   # Feb 11 sorts AFTER Mar 1? Depends on tool.

# BAD: mutable names — what does "latest" point to next week?
orders_latest.parquet
orders_final.parquet
orders_final_v2.parquet
orders_final_v2_ACTUALLY_FINAL.parquet   # this is not a joke

import uuid
from datetime import date

def make_output_filename(
    entity:    str,
    run_date:  date,
    run_id:    str,
    chunk_idx: int,
    fmt:       str = 'parquet',
) -> str:
    return f"{entity}_{run_date.strftime('%Y%m%d')}_run-{run_id[:8]}_{chunk_idx:04d}.{fmt}"

run_id = str(uuid.uuid4())
print(make_output_filename('orders', date(2026, 3, 17), run_id, 1))

orders_20260317_run-f8a3b2c4_0001.parquet

# if a run writes chunks 0001 and 0002 but not 0003, you know from the
# filenames alone that it stopped early — no log file needed to tell you that

# Hive-style partitioning uses key=value directory names.
# Query engines (Spark, Athena, Presto, BigQuery external tables)
# understand this structure natively and prune partitions automatically.

s3://freshcart-lake/silver/orders/
  date=2026-03-15/
    part-00001.parquet    (rows where date = 2026-03-15)
    part-00002.parquet
  date=2026-03-16/
    part-00001.parquet    (rows where date = 2026-03-16)
  date=2026-03-17/
    part-00001.parquet    (rows where date = 2026-03-17)
    part-00002.parquet
    part-00003.parquet

# Multi-level partitioning (for finer granularity):
s3://freshcart-lake/silver/orders/
  year=2026/month=03/day=17/
    store=ST001/part-00001.parquet
    store=ST002/part-00001.parquet

-- Query: SELECT COUNT(*) FROM orders WHERE date = '2026-03-17'
-- Without partitioning:   reads ALL files  → 100% I/O
-- With date partitioning: reads ONLY date=2026-03-17/ → ~0.3% I/O (1 of 365 days)

-- Query: WHERE year=2026 AND month=03 AND store='ST001'
-- Reads: ONLY year=2026/month=03/*/store=ST001/ files

# PyArrow:
import pyarrow as pa
import pyarrow.parquet as pq

df['date'] = pd.to_datetime(df['created_at']).dt.date.astype(str)
table = pa.Table.from_pandas(df)
pq.write_to_dataset(
    table, root_path='s3://freshcart-lake/silver/orders',
    partition_cols=['date'],            # creates date=YYYY-MM-DD/ dirs
    filesystem=s3_filesystem, compression='snappy',
    existing_data_behavior='overwrite_or_ignore',
)

# PySpark:
df.write.mode('overwrite').partitionBy('date').parquet('s3://freshcart-lake/silver/orders')

# SPLITTABLE: Parquet with Snappy, ZSTD, or LZ4
# Each row group is compressed independently — Spark assigns one row
# group per task, so all executors work in parallel.
s3://freshcart-lake/silver/orders/date=2026-03-17/part-00001.parquet

# NON-SPLITTABLE: plain .gz CSV
# One executor must decompress the ENTIRE file before splitting; the
# rest sit idle.
s3://freshcart-lake/landing/orders_20260317.csv.gz

500 MB Parquet + Snappy, 10 row groups, 10 executors:
→ read time: ~5 seconds (true parallel read)

500 MB .csv.gz, same 10 executors:
→ read time: ~50 seconds (one executor decompresses alone, others idle)

# DATA LAKE — Parquet files (Bronze, Silver, Gold)
pq.write_table(table, path, compression='zstd')     # best all-rounder for 2026
pq.write_table(table, path, compression='snappy')   # safe default, widely supported

# Per-column compression (Parquet supports different codecs per column):
pq.write_table(table, path, compression={
    'order_id':   'zstd',    # numeric ID — compresses well with delta encoding
    'order_text': 'snappy',  # free text — fast decomp matters for queries
    'image_url':  'gzip',    # URL strings — ratio more important than speed
})

# KAFKA / STREAMING — LZ4 for lowest latency and minimal CPU overhead
# producer config: compression.type=lz4

# LANDING ZONE — accept whatever the vendor sends as-is, convert at Bronze
# ARCHIVAL (data > 2 years old) — GZIP for CSV, ZSTD level 19 for Parquet

# ZSTD compression level tuning:
pq.write_table(table, path, compression='zstd', compression_level=1)  # fastest (default)
# compression_level=9  → high ratio, slower — for archival
# compression_level=19 → max ratio, very slow — cold storage only

# HEALTHY: few large files per partition
s3://freshcart-lake/silver/orders/date=2026-03-17/
  part-00001.parquet  (480 MB)
  part-00002.parquet  (520 MB)
  part-00003.parquet  (495 MB)
# 3 files, ~500 MB each → 3 Spark tasks, each reads one large file efficiently

# SMALL FILE PROBLEM: many tiny files per partition
s3://freshcart-lake/silver/orders/date=2026-03-17/
  part-00001.parquet  (2.1 KB)   ← written by 5-minute micro-batch 00:05
  part-00002.parquet  (1.8 KB)   ← written by 5-minute micro-batch 00:10
  ... (286 more files) ...
  part-00288.parquet  (1.9 KB)   ← written by 5-minute micro-batch 23:55
# 288 files, ~576 KB total (same data!) → 288 Spark tasks, 288× overhead

S3 LIST calls:       3 files <10ms  •  288 files ~100ms  •  50,000 files 5-10s
Spark task overhead: ~100ms/task × 288 tasks = ~29s of pure scheduling overhead
                      (vs. ~300ms for the 3-file layout — 97× more overhead)
Parquet footer reads: 288 separate S3 GET requests just for file metadata

Scale of the problem: a pipeline writing every 5 minutes across 10 store
partitions produces 2,880 files/day → 1,051,200 files/year. Most systems
start struggling well before 1M files.

from pyspark.sql import SparkSession

def compact_partition(spark: SparkSession, path: str, date: str, target_file_size_mb: int = 512) -> None:
    """Read all small files in a partition and rewrite as fewer large files."""
    partition_path = f"{path}/date={date}"
    df = spark.read.parquet(partition_path)
    row_count = df.count()

    # Rough estimate: 1M rows ≈ 100 MB compressed Parquet
    estimated_mb  = row_count / 10_000
    target_files  = max(1, int(estimated_mb / target_file_size_mb))

    df.coalesce(target_files).write.mode('overwrite').parquet(partition_path)
    print(f"Compacted date={date}: {row_count:,} rows → {target_files} files")

Compacted date=2026-03-17: 48,200,000 rows → 6 files
# 288 tiny files → 6 files of ~500 MB each, same data, same query results

# Coalesce the output DataFrame before writing:
def write_compact_parquet(df, output_path: str, partition_col: str = 'date') -> None:
    row_count    = df.count()
    target_mb    = 512
    rows_per_mb  = 10_000
    target_files = max(1, int(row_count / (target_mb * rows_per_mb)))

    df.repartition(target_files, partition_col) \
      .write.mode('overwrite').partitionBy(partition_col).parquet(output_path)


# Delta Lake / Iceberg have built-in compaction — run on a schedule instead:
# OPTIMIZE silver.orders WHERE date = '2026-03-17';
from delta.tables import DeltaTable
delta_table = DeltaTable.forPath(spark, 's3://freshcart-lake/silver/orders')
delta_table.optimize().where("date = '2026-03-17'").executeCompaction()
# OPTIMIZE does not change table content, only file organisation —
# run VACUUM afterward to reclaim space from the old small files

streamDf.writeStream \
    .trigger(processingTime='1 hour') \
    .option('maxRecordsPerFile', 500_000) \
    .partitionBy('date') \
    .parquet('s3://freshcart-lake/silver/orders')

# processingTime='1 hour' batches an hour of data before writing:
# → 24 files/day instead of 288 files/day (12× improvement)
# maxRecordsPerFile caps the maximum records per output file

# TOO SMALL (< 32 MB): overhead dominates the actual read
#   S3 GET latency:        ~10ms
#   Parquet footer read:   ~5ms
#   Spark task scheduling: ~100ms
#   For a 1 KB file:   overhead (115ms) >> actual read time (<1ms)
#   For a 512 MB file: overhead (115ms) << actual read time (~500ms)
#   Metadata catalog also grows large — one entry per file

# TOO LARGE (> 2 GB): single-executor bottleneck
#   Spark cannot split a single file across tasks by bytes, only by row group
#   One 4 GB file  → one executor reads all 4 GB serially
#   Four 1 GB files → four executors read 1 GB each in parallel
#   A failed write of 4 GB also wastes 4 GB of work; 512 MB wastes far less

# THE SWEET SPOT: 256 MB – 1 GB per file (compressed Parquet)
# minimum 128 MB, maximum 1-2 GB

import pyarrow.parquet as pq

metadata = pq.read_metadata('s3://freshcart-lake/silver/orders/date=2026-03-17/part-00001.parquet')
file_size_mb   = metadata.serialized_size / (1024 * 1024)
row_count      = metadata.num_rows
num_row_groups = metadata.num_row_groups

print(f"File size: {file_size_mb:.1f} MB")
print(f"Row count: {row_count:,}")
print(f"Row groups: {num_row_groups}")
print(f"MB per million rows: {file_size_mb / row_count * 1_000_000:.1f}")

File size: 487.3 MB
Row count: 3,842,900
Row groups: 8
MB per million rows: 126.8
# use this last number to size target_files in the compaction functions above

pq.write_table(
    table, path,
    row_group_size=500_000,   # 500k rows per row group
)

pq.write_table(
    table, 'output.parquet', compression='zstd',
    write_bloom_filter=True,
    bloom_filter_columns=['payment_id', 'order_id'],   # high-cardinality point-lookup columns
    bloom_filter_false_positive_rate=0.05,              # lower rate → larger filter, better pruning
)

# Snowflake / BigQuery achieve a similar effect via table clustering:
# ALTER TABLE silver.orders CLUSTER BY (date, store_id);

import logging
import chardet
import pandas as pd
import pyarrow as pa
from typing import Iterator

log = logging.getLogger('csv_to_parquet')

# Explicit schema — better than letting Parquet infer types from messy CSV
ORDERS_SCHEMA = pa.schema([
    pa.field('order_id',    pa.int64(),                  nullable=False),
    pa.field('store_id',    pa.string(),                  nullable=False),
    pa.field('customer_id', pa.int64(),                  nullable=True),
    pa.field('amount',      pa.decimal128(10, 2),        nullable=False),
    pa.field('status',      pa.string(),                  nullable=False),
    pa.field('created_at',  pa.timestamp('us', tz='UTC'), nullable=False),
    pa.field('ingested_at', pa.timestamp('us', tz='UTC'), nullable=False),
])


def detect_encoding(filepath: str, sample_bytes: int = 100_000) -> str:
    """Detect file encoding from the first N bytes."""
    with open(filepath, 'rb') as f:
        raw = f.read(sample_bytes)
    result = chardet.detect(raw)
    encoding, confidence = result.get('encoding') or 'utf-8', result.get('confidence', 0)
    log.info(f'Detected encoding: {encoding} (confidence: {confidence:.0%})')
    return encoding if confidence > 0.7 else 'utf-8'


def read_csv_chunked(filepath: str, chunk_size: int = 200_000) -> Iterator[pd.DataFrame]:
    """Read CSV in chunks, with detected encoding and bad-line tolerance."""
    encoding = detect_encoding(filepath)
    try:
        for chunk in pd.read_csv(
            filepath, chunksize=chunk_size, encoding=encoding, encoding_errors='replace',
            dtype=str, na_values=['', 'NULL', 'null', 'N/A', 'n/a', 'NA', '-'],
            on_bad_lines='warn',
        ):
            yield chunk
    except Exception as e:
        log.error('Failed to read CSV %s: %s', filepath, str(e))
        raise

from datetime import date

def cast_chunk(chunk: pd.DataFrame, source_date: date) -> pd.DataFrame:
    """Apply type casting and add pipeline metadata columns."""
    df = chunk.copy()

    for col in ['order_id', 'customer_id']:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors='coerce').astype('Int64')

    if 'amount' in df.columns:
        df['amount'] = pd.to_numeric(df['amount'].str.replace(',', '', regex=False), errors='coerce').round(2)

    if 'created_at' in df.columns:
        df['created_at'] = pd.to_datetime(df['created_at'], utc=True, errors='coerce')

    df['ingested_at'] = pd.Timestamp.now(tz='UTC')
    df['source_date']  = source_date.isoformat()   # partition column

    bad_mask = df['order_id'].isna()
    if bad_mask.any():
        log.warning('Dropping %d rows with NULL order_id', bad_mask.sum())
        df = df[~bad_mask]

    return df

import pyarrow.parquet as pq
import pyarrow.fs as pafs
from pathlib import Path

def convert_csv_to_parquet(input_path: str, output_root: str, source_date: date, s3_bucket: str | None = None) -> dict:
    """Convert a CSV file to partitioned Parquet. Returns rows_written/rows_rejected/files_written."""
    stats = {'rows_written': 0, 'rows_rejected': 0, 'files_written': 0}
    all_tables = []

    for chunk_idx, chunk in enumerate(read_csv_chunked(input_path)):
        original_rows = len(chunk)
        cast    = cast_chunk(chunk, source_date)
        dropped = original_rows - len(cast)
        if dropped:
            stats['rows_rejected'] += dropped
            log.warning('Chunk %d: dropped %d invalid rows', chunk_idx, dropped)

        if len(cast) > 0:
            try:
                all_tables.append(pa.Table.from_pandas(cast, schema=ORDERS_SCHEMA, safe=False))
            except Exception as e:
                log.error('Schema cast failed on chunk %d: %s', chunk_idx, str(e))
                raise
        stats['rows_written'] += len(cast)

    if not all_tables:
        log.warning('No valid rows to write')
        return stats

    full_table  = pa.concat_tables(all_tables)
    output_path = f"{s3_bucket}/bronze/orders" if s3_bucket else output_root + "/orders"
    filesystem  = pafs.S3FileSystem(region='ap-south-1') if s3_bucket else pafs.LocalFileSystem()
    if not s3_bucket:
        Path(output_path).mkdir(parents=True, exist_ok=True)

    pq.write_to_dataset(
        full_table, root_path=output_path, partition_cols=['source_date'],
        filesystem=filesystem, compression='zstd', row_group_size=500_000,
        write_statistics=True, existing_data_behavior='overwrite_or_ignore',
    )

    log.info(f"Conversion complete: {stats['rows_written']:,} rows written, "
             f"{stats['rows_rejected']:,} rejected, source_date={source_date.isoformat()}")
    return stats

INFO Detected encoding: utf-8 (confidence: 99%)
WARNING Chunk 3: dropped 12 invalid rows
INFO Conversion complete: 199,988 rows written, 12 rejected, source_date=2026-03-17

import json
from typing import Iterator

def read_ndjson(filepath: str) -> Iterator[dict]:
    """Stream records from an NDJSON file one at a time."""
    with open(filepath, encoding='utf-8') as f:
        for line_num, line in enumerate(f, start=1):
            line = line.strip()
            if not line:
                continue
            try:
                yield json.loads(line)
            except json.JSONDecodeError as e:
                log.warning('Invalid JSON on line %d: %s', line_num, str(e))


def flatten_order(raw: dict) -> dict:
    """Flatten nested order JSON to a flat dict for Parquet storage."""
    return {
        'order_id':        raw.get('order_id') or raw.get('id'),
        'customer_id':     raw.get('customer', {}).get('id'),
        'customer_city':   raw.get('customer', {}).get('address', {}).get('city'),
        'restaurant_id':   raw.get('restaurant', {}).get('id'),
        'order_amount':    raw.get('payment', {}).get('amount'),
        'payment_method':  raw.get('payment', {}).get('method'),
        'item_count':      len(raw.get('items', [])),
        'status':          raw.get('status'),
        'created_at':      raw.get('created_at'),
        '_raw_items':      json.dumps(raw.get('items', [])),   # kept for reference
    }

import pyarrow.parquet as pq

def json_to_parquet(input_path: str, output_path: str, batch_size: int = 100_000) -> int:
    """Convert NDJSON to Parquet with flattening. Returns total rows written."""
    total, batch, writer = 0, [], None

    for record in read_ndjson(input_path):
        batch.append(flatten_order(record))
        if len(batch) >= batch_size:
            table = pa.Table.from_pandas(pd.DataFrame(batch))
            if writer is None:
                writer = pq.ParquetWriter(output_path, table.schema, compression='zstd')
            writer.write_table(table)
            total += len(batch)
            batch = []

    if batch:
        table = pa.Table.from_pandas(pd.DataFrame(batch))
        if writer is None:
            writer = pq.ParquetWriter(output_path, table.schema, compression='zstd')
        writer.write_table(table)
        total += len(batch)

    if writer:
        writer.close()
    log.info(f"Wrote {total:,} rows to {output_path}")
    return total

WARNING Invalid JSON on line 40218: Expecting ',' delimiter: line 1 column 812
INFO Wrote 199,999 rows to s3://freshcart-lake/bronze/orders_json/2026-03-17.parquet

resource "aws_s3_bucket_lifecycle_configuration" "freshcart_lake" {
  bucket = "freshcart-data-lake"

  # Landing zone — raw files only needed until converted to Bronze
  rule {
    id     = "landing-zone-cleanup"
    status = "Enabled"
    filter { prefix = "landing/" }
    transition { days = 7  storage_class = "STANDARD_IA" }
    expiration { days = 30 }
  }

  # Bronze — keep 1 year in Standard, then archive
  rule {
    id     = "bronze-archive"
    status = "Enabled"
    filter { prefix = "bronze/" }
    transition { days = 90  storage_class = "STANDARD_IA" }
    transition { days = 365 storage_class = "GLACIER_IR" }
  }
}

resource "aws_s3_bucket_lifecycle_configuration" "freshcart_lake_silver" {
  bucket = "freshcart-data-lake"

  rule {
    id     = "silver-lifecycle"
    status = "Enabled"
    filter { prefix = "silver/" }
    transition { days = 180 storage_class = "STANDARD_IA" }
    transition { days = 730 storage_class = "GLACIER" }   # 2 years
  }

  # Gold layer — usually small, kept hot forever, no rule needed
}

-- Default retention: 7 days (safe minimum for active transactions)
VACUUM silver.orders RETAIN 168 HOURS;

-- Python API, with a dry run first:
from delta.tables import DeltaTable
dt = DeltaTable.forPath(spark, 's3://freshcart-lake/silver/orders')
dt.vacuum(retentionHours=168, dry_run=True)   # shows what WOULD be deleted
dt.vacuum(retentionHours=168)                  # then actually delete

# Schedule weekly on all Silver/Gold tables: 0 3 * * 0 (Sunday 3 AM)

import boto3
from collections import defaultdict

def audit_s3_prefix(bucket: str, prefix: str) -> dict:
    """Count files, total size, and flag small-file-problem symptoms."""
    s3 = boto3.client('s3')
    paginator = s3.get_paginator('list_objects_v2')
    sizes, partition_files = [], defaultdict(list)

    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        for obj in page.get('Contents', []):
            size_mb = obj['Size'] / (1024 * 1024)
            sizes.append(size_mb)
            partition = '/'.join(obj['Key'].split('/')[:-1])
            partition_files[partition].append(size_mb)

    if not sizes:
        return {'error': 'No files found'}

    small_files = sum(1 for s in sizes if s < 10)
    report = {
        'total_files': len(sizes), 'total_gb': round(sum(sizes) / 1024, 2),
        'avg_mb': round(sum(sizes) / len(sizes), 1),
        'small_files': small_files, 'small_pct': round(small_files / len(sizes) * 100, 1),
        'partitions': len(partition_files),
    }
    if small_files / len(sizes) > 0.5:
        report['warning'] = 'More than 50% of files are < 10 MB — compaction needed'
    return report

{
  'total_files': 48234, 'total_gb': 142.7, 'avg_mb': 3.0,
  'small_files': 45891, 'small_pct': 95.1, 'partitions': 1095,
  'warning': 'More than 50% of files are < 10 MB — compaction needed'
}
# avg_mb of 3.0 against a 128-512 MB target, with 95% of files under 10 MB,
# is exactly the small file problem from Part 05 — this audit is what catches it

An analyst runs a monthly revenue report every Monday morning. It used to take 90 seconds. This week it took 18 minutes. Nothing changed in the query. You investigate.

# Step 1: Athena console → query execution detail
# Data scanned: 4.2 TB  ← the problem signal (expected ~400 GB for 90 days)

# Step 2: check partition count in the catalog
$ aws glue get-partitions --database-name freshcart_silver --table-name events \
    --query 'Partitions | length(@)'
# 156,420 partitions — expected ~1,095 for 3 years of daily data

# Step 3: find the culprit partition key
$ aws glue get-partitions --database-name freshcart_silver --table-name events \
    --query 'Partitions[0:5].StorageDescriptor.Location'
# s3://freshcart-lake/silver/events/date=2026-03-17/hour=20/minute=14/
# s3://freshcart-lake/silver/events/date=2026-03-17/hour=20/minute=15/
# → the pipeline was accidentally partitioning by MINUTE

$ aws s3 ls s3://freshcart-lake/silver/events/ --recursive | awk '{print $4}' | wc -l
# 4,847,293 files — each ~1 KB (one minute of events)

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()

df = spark.read.parquet('s3://freshcart-lake/silver/events/')
df.write.mode('overwrite').partitionBy('date').parquet('s3://freshcart-lake/silver/events_v2/')

spark.sql('MSCK REPAIR TABLE freshcart_silver.events_v2')

Before → After
Partitions:      156,420  →  1,095
Files:            ~4.8M   →  ~3,285
Avg file size:    ~1 KB   →  ~130 MB
Query time:      18 min   →  94 sec   (11× improvement)
Data scanned:     4.2 TB  →  412 GB   (10× reduction)

The root cause was a single line in the Spark write configuration that added minute as a partition column alongside date. This turned 1,095 daily partitions into 1.5 million minute partitions — all valid, all correct data, but completely unusable for analytics. Two hours of investigation and a Spark rewrite job fixed it permanently.