:::tip 🎮 Interactive Playground Visualize this concept: Try the Storage Formats Compared demo on the EngineersOfAI Playground - no code required. :::

Storage Formats for ML Training Data

The $47,000 CSV File

It was a routine model retraining job. Every Sunday at 2 AM, the feature pipeline at a mid-size fintech company read 18 months of transaction history, computed 200+ features, and wrote a training dataset for the fraud detection model. The data science team had set it up eight months ago and it mostly worked. It just took a while.

A while meant 11 hours. The pipeline read a 340 GB CSV file from S3, joined it against several dimension tables, computed aggregations, and wrote out a training-ready file. For 11 hours every Sunday, a six-node Spark cluster hummed along at full utilization. The team had accepted this as the cost of doing ML.

When a new data engineer joined the team and asked why the training file was stored as CSV, the answer was "that's what the data science team was comfortable with." It was an honest answer. The original data scientists had exported the data as CSV because pandas reads CSV. Nobody had revisited the decision as the dataset grew from 10 GB to 340 GB.

The data engineer converted the pipeline to Parquet in an afternoon. The training job now read only the 47 columns the model actually used out of 200+, with pushdown predicates to skip 14 months of data that wasn't in the training window. The 11-hour job became 38 minutes. The Spark cluster shrank from six nodes to two. The annual compute savings were approximately $47,000.

The CSV file wasn't the bottleneck because CSV is a bad format for storing comma-separated values. It's a perfectly reasonable format for that. It was the bottleneck because the access pattern - read 47 columns out of 200, filter by date range - is precisely the access pattern that CSV is worst at and Parquet is best at. Storage format choice is not an abstract engineering preference. It's a direct lever on pipeline cost and training iteration speed.

This lesson explains exactly why that's true, and how to think clearly about which format to use for every stage of your ML data pipeline.

---

Why This Exists

The original data storage format on disk was row-oriented: a record is a contiguous sequence of bytes representing all fields of one row. This is how CSV, TSV, and most relational databases store data on disk. It maps naturally to how humans think about records - a row is a thing, and you store all its properties together.

Row orientation is optimal for transactional workloads. When you insert, update, or read a single record by primary key, you read one contiguous chunk of disk and get all the fields. Random access is fast. Write patterns are simple. This is why PostgreSQL, MySQL, and almost all OLTP databases are row-oriented.

Analytical workloads have a fundamentally different access pattern. Instead of reading one record with all fields, an analytical query reads all values of a few fields across many records. SELECT avg(amount) FROM transactions WHERE merchant_category = 'food' needs the amount and merchant_category columns across millions of rows. It doesn't need user_id, device_type, ip_address, or any of the other 50 columns in the table.

In a row-oriented format, reading two columns still requires reading every byte of every row - all 52 columns of data pass through the I/O system even though 50 columns are discarded immediately. At 100M rows, this is the difference between reading 4 GB (two columns, compressed) and reading 200 GB (all 52 columns).

Columnar storage was invented to fix this. By storing all values of a column contiguously on disk, a columnar format lets you read exactly the columns you need. The physical I/O exactly matches the logical query. This single insight - columns together, rows apart - drives the entire modern data infrastructure stack.

---

Historical Context

The theoretical case for columnar storage goes back to academic database research in the 1970s, but practical implementations arrived much later. The first widely-used columnar database was Sybase IQ (1994), which demonstrated that analytical queries could be 10-100x faster on columnar storage.

The open-source columnar format era started with Google's Dremel paper (2010), which described the columnar storage format used internally at Google for petabyte-scale analytics. Dremel introduced the repetition and definition levels encoding that handles nested data structures - the same encoding that Apache Parquet later adopted directly.

Parquet was created by Twitter and Cloudera in 2013 as an open-source implementation of the Dremel columnar format for the Hadoop ecosystem. The first release landed in April 2013. It was accepted into the Apache Software Foundation and became a top-level project in 2015. Parquet is now effectively the universal format for big data analytics.

Apache ORC (Optimized Row Columnar) was created by Hortonworks and Facebook in 2013, also for Hadoop. ORC added built-in ACID transaction support and was designed to be the native format for Hive. For a period, Parquet (Spark-native) and ORC (Hive-native) competed; today Parquet has won the broader ecosystem, though ORC remains relevant in Hive-heavy shops.

Apache Avro originated at Hadoop (2009) with a different goal: not query performance but schema evolution. Avro is row-oriented, with a binary encoding and a separate JSON schema. It was designed for data serialization and streaming - Kafka heavily influenced Avro's adoption as the de facto streaming serialization format.

The table format layer - Delta Lake (Databricks, 2019), Apache Iceberg (Netflix, open-sourced 2018), and Apache Hudi (Uber, 2019) - emerged to solve a different problem: ACID transactions and time travel on top of Parquet files stored in object storage. These are not storage formats themselves but metadata layers that give Parquet files database-like properties.

---

Core Concepts

Row-Oriented vs. Columnar: The Fundamental Trade-off

Imagine a table with 1 billion rows and 100 columns. You run an analytical query that needs 5 columns.

Row-oriented storage (CSV, JSON):

The data on disk looks like:

[row1_col1][row1_col2]...[row1_col100] | [row2_col1]...[row2_col100] | ...

To read 5 columns, you must read all 100 columns for every row. If each column averages 8 bytes:

• Bytes needed: $5 \times 8 \times 10^9 = 40 \text{ GB}$
• Bytes actually read: $100 \times 8 \times 10^9 = 800 \text{ GB}$
• I/O amplification factor: $\frac{800}{40} = 20\times$

Columnar storage (Parquet, ORC):

The data on disk looks like:

[all row1..rowN col1 values] | [all row1..rowN col2 values] | ...

To read 5 columns, you read only those 5 columns:

• Bytes read: $5 \times 8 \times 10^9 = 40 \text{ GB}$
• I/O amplification factor: $1\times$

This is the column pruning benefit. At 20x I/O reduction, a job that took 11 hours now takes 33 minutes - roughly what the fintech team observed.

Columnar storage also enables better compression. Within a column, all values are the same type. Values in a column are often correlated - a country_code column has only ~200 distinct values across billions of rows. This homogeneity allows aggressive compression:

• Dictionary encoding: store unique values once, store integer indices in the data
• Run-length encoding: store repeated values as (value, count) pairs
• Delta encoding: store differences between consecutive values (ideal for timestamps)

A typical mixed-type CSV compresses at 3-5x with gzip. A typed Parquet file with dictionary encoding compresses at 8-20x. Real-world compression ratios depend heavily on data characteristics, but columnar generally wins by a factor of 2-4x on compression ratio alone.

Predicate Pushdown

Predicate pushdown means filtering data at the storage layer before it reaches the compute layer. Instead of reading all data and then filtering in memory, the storage format stores metadata that lets the query engine skip entire chunks of data that can't possibly contain matching rows.

Parquet stores column statistics (min, max, null count) for each row group and each column chunk. When a query has a predicate like WHERE transaction_date >= '2024-01-01', the query engine reads the row group metadata first, checks the transaction_date max value for each row group, and skips any row group where max(transaction_date) < '2024-01-01'. No data is read for those row groups.

If your data is sorted or clustered by transaction_date (a common pattern for time-series data), predicate pushdown can skip 90%+ of the file. The effective I/O is proportional to the selectivity of the predicate, not the size of the full dataset.

For ML pipelines that train on a rolling time window (e.g., the last 90 days of a 2-year history), predicate pushdown on the timestamp column is a first-order optimization. This alone can reduce I/O by 5-10x on time-partitioned datasets.

Vectorized Reads

Modern CPUs execute SIMD (Single Instruction Multiple Data) operations that apply one instruction to a vector of values simultaneously. A 256-bit AVX instruction can operate on 32 bytes (eight 32-bit integers) in a single clock cycle.

Columnar formats enable vectorized reads because all values in a column chunk are the same type, laid out contiguously in memory. The CPU can process them in batches without branching on type or skipping over unrelated fields. PyArrow and DuckDB both use SIMD-accelerated columnar processing internally - this is why DuckDB can aggregate a billion rows locally faster than many distributed Spark clusters.

Row-oriented formats break vectorization: adjacent bytes in memory are from different columns with different types. The processing loop must handle each row individually, defeating SIMD optimization.

---

Parquet Deep Dive

Parquet is the most important format to understand deeply because it underpins virtually all modern ML data pipelines. Every feature store, every lakehouse, every ML training pipeline you'll encounter at scale uses Parquet.

Internal Structure

A Parquet file has three levels of structure:

Row Groups: A Parquet file is divided into horizontal partitions called row groups. The default row group size is 128 MB. Each row group is stored contiguously on disk and can be read independently. Row groups are the unit of parallelism - one Spark task or one PyArrow read thread processes one row group.

Column Chunks: Within each row group, each column's data is stored as a column chunk. A column chunk is a contiguous block of bytes containing only values from one column for one row group.

Pages: Column chunks are subdivided into pages (typically 1 MB). Pages are the unit of compression and encoding. Two types of pages matter:

• Data pages: the actual encoded column values
• Dictionary pages: when dictionary encoding is active, the dictionary of unique values is stored in a separate page at the start of the column chunk

File Footer: The last section of the Parquet file is the footer - a Thrift-encoded metadata block containing:

• Schema (column names, types, logical types)
• Row group metadata (offset, size, row count)
• Column chunk metadata (offset, encoding, compression codec)
• Column statistics (min, max, null count) per column chunk and per row group

The query engine reads the footer first (a small read, typically under 1 MB) to build a complete index of the file. It then reads only the row groups and column chunks needed by the query. This is why predicate pushdown works: the engine never reads data it knows can't match.

Row Group Size Tuning

The default 128 MB row group size is a reasonable starting point. For ML pipelines, tuning it matters:

Larger row groups (256 MB - 1 GB):

• Better compression (more data for dictionary encoding to work with)
• Better column statistics (statistics represent more data, better pushdown)
• Worse parallelism (fewer row groups = fewer parallel tasks)
• Higher memory requirement (must buffer the whole row group to write)

Smaller row groups (32 MB - 64 MB):

• Better parallelism on distributed systems (more tasks)
• Lower memory requirement during write
• Worse compression and statistics
• Higher overhead (more footer metadata)

For ML training data that will be read by many parallel workers (Spark, Ray, PyTorch DataLoader with multiple workers), smaller row groups (64-128 MB) enable better parallelism. For archival storage that will be queried with selective predicates, larger row groups (256 MB+) improve compression and predicate pushdown effectiveness.

Bloom Filters in Parquet 2.0

Min/max statistics work well for range predicates (WHERE date >= X). They don't help for equality predicates on high-cardinality columns (WHERE user_id = 'abc123'). If user IDs are uniformly distributed, the min/max of every row group spans the entire keyspace - statistics give no information about whether a specific user_id is present.

Parquet 2.0 introduced bloom filters to solve this. A bloom filter is a probabilistic data structure that answers "is this value definitely absent from this row group?" with a false positive rate (typically 1%). If the bloom filter says a value is absent, the row group can be skipped with certainty. If it says the value might be present, the row group must be read.

Bloom filters add storage overhead (proportional to the number of distinct values and the false positive rate) but can dramatically reduce I/O for point lookups on high-cardinality keys. For ML pipelines that filter training data by entity ID (e.g., "all transactions from users in the test set"), bloom filters on the entity_id column can eliminate most I/O.

Dictionary Encoding

When a column has many repeated values (low cardinality), dictionary encoding replaces each value with a small integer index into a per-column-chunk dictionary of unique values. For a country_code column with 50 distinct values across 10 million rows, dictionary encoding stores 50 strings once and 10 million 6-bit integers. This is dramatically more compact than storing 10 million full strings.

Dictionary encoding has a failure mode: dictionary explosion. When a column has too many distinct values (high cardinality), the dictionary grows too large and Parquet falls back to plain encoding. The cardinality threshold depends on value size; for strings, dictionaries typically work well up to ~10,000 distinct values per row group. Beyond that, expect plain encoding with correspondingly worse compression.

For ML feature data: categorical features (country, device type, product category) benefit enormously from dictionary encoding. High-cardinality IDs (user_id, transaction_id, session_id) stored as strings will cause dictionary explosion - store them as integers where possible, or accept plain encoding.

---

Avro: Schema Evolution First

Apache Avro is a row-oriented binary serialization format with a key design distinction: the schema is always stored alongside the data (either embedded or referenced via a schema registry). Avro was designed specifically for:

1. Schema evolution: backward and forward compatibility between schema versions
2. Streaming data: efficient serialization for high-throughput message queues (Kafka)
3. Language interoperability: schemas defined in JSON, readable by any language

Avro's schema evolution rules:

• Adding a field with a default value: backward compatible (old readers ignore new fields; new readers use defaults for old data)
• Removing a field with a default value: forward compatible
• Renaming a field: breaking change - requires an alias or a new field

For ML pipelines, Avro is the right choice when:

• Data arrives via Kafka or other message queues and is written as streaming events
• Schema evolution is a primary concern (you expect to add/remove fields frequently)
• You need to read the data in languages where Parquet support is weaker

Avro is the wrong choice for ML training data reads because it's row-oriented - it has none of Parquet's column pruning or predicate pushdown benefits. The typical pattern: write events to Kafka as Avro, compact them to Parquet daily for training data consumption.

---

ORC: Hive-Native Columnar

ORC (Optimized Row Columnar) was developed by Facebook and Hortonworks for Hive. It predates Parquet by a few months. ORC and Parquet have similar columnar architectures; the differences are:

• ACID support: ORC has built-in ACID transaction support for Hive (insert, update, delete with transaction semantics). Parquet has no native ACID support.
• Predicate pushdown: Both support it; ORC's implementation is slightly more mature for Hive use cases.
• Stripe vs. Row Group: ORC calls its horizontal partitions "stripes" (same concept as row groups, default 64 MB).
• Built-in indexes: ORC includes bloom filters and row-level indexes by default; Parquet added bloom filters later (v2.0).
• Ecosystem: ORC is the native format for Hive and preferred by many Hadoop shops. Parquet is the native format for Spark and the broader ecosystem.

For ML pipelines on AWS, GCP, or Azure: use Parquet unless you're in a Hive-heavy environment where ORC is already established.

---

Delta Lake, Apache Iceberg, and Apache Hudi

Parquet files on object storage (S3, GCS, ADLS) have a fundamental problem: they provide no ACID guarantees. If two writers attempt to append to the same "table" simultaneously, you get data corruption. There's no atomic commit, no transaction isolation, no way to roll back a failed write.

The open table formats (Delta Lake, Iceberg, Hudi) solve this by adding a metadata layer on top of Parquet files. The actual data is still stored as Parquet. The table format adds:

1. ACID transactions: atomic commits, serializable isolation
2. Table history: every change creates a new snapshot; old snapshots are retained
3. Time travel: query the table as it existed at any past timestamp or version
4. Schema evolution: tracked and enforced at the metadata layer
5. Partition evolution: change partitioning strategy without rewriting all data
6. Compaction: merge small files into larger ones without downtime

Delta Lake

Delta Lake (Databricks, open-sourced 2019) stores a transaction log as a sequence of JSON files in a _delta_log/ directory alongside the Parquet data files. Each commit adds a new JSON log entry describing the operation (add files, remove files, update metadata). Reading the log gives you the complete table state at any version.

Delta Lake is the most widely adopted open table format, driven by Databricks' market presence and deep Spark integration.

Apache Iceberg

Iceberg (Netflix, open-sourced 2018 as Apache project) takes a more principled approach. The metadata layer uses a hierarchy: catalog → table metadata file → snapshot → manifest list → manifest files → data files. This design enables better performance for tables with millions of partitions and supports multiple query engines (Spark, Flink, Trino, Dremio) without engine-specific dependencies.

Iceberg has stronger support for partition evolution - you can change how a table is partitioned without rewriting data, which is a major operational advantage.

Apache Hudi

Hudi (Uber, 2019) focuses on incremental data pipelines and near-real-time analytics. Its key differentiators are built-in support for upserts (update-or-insert) and incremental queries - "give me all records that changed since version N." This makes Hudi natural for CDC (Change Data Capture) pipelines where you're continuously applying database change events to an analytical table.

For ML, all three formats offer the same core value: time travel for point-in-time training data. Instead of maintaining a separate SCD Type 2 snapshot table, you can query your Delta/Iceberg/Hudi table AS OF a specific timestamp, and the table format reconstructs what the data looked like at that moment.

---

Compression Codecs

Compression reduces storage cost and improves I/O throughput (less data to read from disk or network). The trade-off is CPU time for compression/decompression. The right codec depends on whether your workload is I/O-bound or CPU-bound.

Codec	Compression Ratio	Decompress Speed	Compress Speed	Best For
Snappy	Moderate (2-4x)	Very fast	Very fast	Streaming, low-latency reads
LZ4	Moderate (2-4x)	Fastest	Fast	Real-time, memory-mapped reads
Gzip	Good (4-8x)	Moderate	Slow	Archival, storage-heavy
Zstd	Very good (4-10x)	Fast	Moderate	Best overall for analytical
Brotli	Excellent (5-12x)	Moderate	Very slow	Static content, one-time write
Uncompressed	1x	Instant	None	Testing only

For ML training data (read many times, written once):

• Zstd is the modern default - excellent ratio with fast decompression. PyArrow and Parquet support ZSTD with configurable compression levels (1-22; level 3 is a good balance).
• Snappy is reasonable if you need maximum read throughput and storage cost is secondary. Common in environments where decompression latency matters (streaming reads, GPUs reading data faster than CPU can decompress with Zstd).
• Gzip is the legacy default for CSV files. Reasonable compression, slow for parallel reads because gzip files aren't splittable - a single gzip file can only be read by one process. Use Parquet + Zstd instead.

The math on Zstd vs. Snappy: if your storage bandwidth is the bottleneck (cloud object storage is typically 500 MB/s - 1 GB/s per VM), better compression means less time waiting for I/O. If your CPU is the bottleneck (e.g., GPU training with minimal preprocessing), faster decompression wins. In most ML training scenarios, I/O is the bottleneck and Zstd wins.

---

Code Examples

Reading and Writing Parquet with PyArrow

import pyarrow as pa
import pyarrow.parquet as pq
import pyarrow.compute as pc
import numpy as np
import pandas as pd
from pathlib import Path

# ── Create sample ML training data ────────────────────────────────────────
np.random.seed(42)
N = 1_000_000  # 1M rows

df = pd.DataFrame({
    "transaction_id":    range(N),
    "user_id":           np.random.randint(1, 50_000, N),
    "merchant_id":       np.random.randint(1, 5_000, N),
    "transaction_date":  pd.date_range("2023-01-01", periods=N, freq="1min"),
    "amount":            np.random.lognormal(4, 1.5, N).round(2),
    "country_code":      np.random.choice(["US", "GB", "DE", "FR", "JP"], N),
    "device_type":       np.random.choice(["mobile", "desktop", "tablet"], N, p=[0.6, 0.3, 0.1]),
    "is_fraud":          np.random.binomial(1, 0.02, N).astype(np.int8),
    # Simulate 200 feature columns
    **{f"feature_{i:03d}": np.random.randn(N).astype(np.float32)
       for i in range(200)},
})

# Convert to PyArrow Table for fine-grained control
table = pa.Table.from_pandas(df, preserve_index=False)

print(f"Table schema (first 10 fields):")
for i, field in enumerate(table.schema):
    if i >= 10:
        print(f"  ... and {len(table.schema) - 10} more fields")
        break
    print(f"  {field.name}: {field.type}")

# ── Write Parquet with tuned settings ─────────────────────────────────────
output_path = "/tmp/training_data.parquet"

pq.write_table(
    table,
    output_path,
    compression="zstd",              # Best ratio + fast decompression
    compression_level=3,             # Level 3: good balance, 1=fast, 22=best ratio
    row_group_size=500_000,          # 500K rows per row group (tunes parallelism)
    # write_statistics=True,         # Default: True - enables predicate pushdown
    use_dictionary=True,             # Default: True - enables dictionary encoding
    dictionary_pagesize_limit=2 * 1024 * 1024,  # 2MB dictionary limit per column
    write_page_index=True,           # Column index for more granular pushdown (Parquet 2.0)
    # Bloom filters (Parquet 2.0) - useful for high-cardinality key columns
    # bloom_filter_columns=["user_id", "merchant_id"],
    # bloom_filter_false_positive_rate=0.01,
)

import os
file_size_mb = os.path.getsize(output_path) / (1024**2)
print(f"\nParquet file written: {file_size_mb:.1f} MB")

# ── Partitioned write (partition by date for time-range queries) ───────────
output_dir = "/tmp/training_partitioned"

# Add partition column
df["year_month"] = df["transaction_date"].dt.to_period("M").astype(str)
table_with_partition = pa.Table.from_pandas(df, preserve_index=False)

pq.write_to_dataset(
    table_with_partition,
    root_path=output_dir,
    partition_cols=["year_month"],
    compression="zstd",
    compression_level=3,
    existing_data_behavior="overwrite_or_ignore",
)
print(f"\nPartitioned dataset written to {output_dir}/")

# ── Reading with column pruning and predicate pushdown ────────────────────
# Only read 5 columns - column pruning at the storage layer
selected_columns = ["transaction_id", "user_id", "amount", "is_fraud", "country_code"]

# Read entire file, pruned columns
table_pruned = pq.read_table(output_path, columns=selected_columns)
print(f"\nColumn-pruned read: {table_pruned.shape} (only {len(selected_columns)} columns)")

# Read with predicate pushdown - only rows where amount > 1000
# PyArrow evaluates this against row group statistics first,
# skipping row groups where max(amount) <= 1000
filters = [("amount", ">", 1000.0)]
table_filtered = pq.read_table(
    output_path,
    columns=selected_columns,
    filters=filters,
)
print(f"Predicate pushdown read: {table_filtered.shape}")
print(f"Rows with amount > 1000: {(df['amount'] > 1000).sum()} (verification)")

# ── Read metadata without reading data ────────────────────────────────────
parquet_file = pq.ParquetFile(output_path)
metadata = parquet_file.metadata

print(f"\nFile metadata:")
print(f"  Row groups:  {metadata.num_row_groups}")
print(f"  Total rows:  {metadata.num_rows:,}")
print(f"  Serialized size: {metadata.serialized_size:,} bytes (footer only)")

for i in range(min(2, metadata.num_row_groups)):
    rg = metadata.row_group(i)
    print(f"  Row group {i}: {rg.num_rows:,} rows, "
          f"{rg.total_byte_size / 1e6:.1f} MB compressed")

Benchmarking CSV vs. Parquet with DuckDB

import duckdb
import pandas as pd
import numpy as np
import time
import os

# ── Generate a moderately sized dataset ──────────────────────────────────
np.random.seed(0)
N = 5_000_000

df = pd.DataFrame({
    "transaction_id":    range(N),
    "user_id":           np.random.randint(1, 100_000, N),
    "amount":            np.random.lognormal(4, 1.5, N).round(2),
    "country_code":      np.random.choice(["US", "GB", "DE", "FR", "JP", "CA", "AU"], N),
    "transaction_date":  pd.date_range("2022-01-01", periods=N, freq="1min"),
    "is_fraud":          np.random.binomial(1, 0.02, N).astype(int),
    **{f"extra_col_{i}": np.random.randn(N).astype(np.float32) for i in range(50)},
})

csv_path     = "/tmp/bench_data.csv"
parquet_path = "/tmp/bench_data.parquet"

# Write CSV
t0 = time.perf_counter()
df.to_csv(csv_path, index=False)
csv_write_s = time.perf_counter() - t0
csv_size_mb = os.path.getsize(csv_path) / (1024**2)

# Write Parquet
t0 = time.perf_counter()
df.to_parquet(parquet_path, compression="zstd", engine="pyarrow")
parquet_write_s = time.perf_counter() - t0
parquet_size_mb = os.path.getsize(parquet_path) / (1024**2)

print(f"File sizes:")
print(f"  CSV:     {csv_size_mb:>8.1f} MB  (write: {csv_write_s:.1f}s)")
print(f"  Parquet: {parquet_size_mb:>8.1f} MB  (write: {parquet_write_s:.1f}s)")
print(f"  Compression ratio: {csv_size_mb / parquet_size_mb:.1f}x")

# ── Benchmark analytical queries ─────────────────────────────────────────
conn = duckdb.connect()

QUERY = """
    SELECT
        country_code,
        COUNT(*)           AS n_transactions,
        SUM(amount)        AS total_amount,
        AVG(amount)        AS avg_amount,
        SUM(is_fraud)      AS n_fraud,
        AVG(is_fraud)      AS fraud_rate
    FROM '{path}'
    WHERE transaction_date >= '2023-06-01'
    GROUP BY country_code
    ORDER BY total_amount DESC
"""

def benchmark(path: str, label: str, n_runs: int = 3) -> float:
    times = []
    for _ in range(n_runs):
        t0 = time.perf_counter()
        conn.execute(QUERY.format(path=path)).fetchdf()
        times.append(time.perf_counter() - t0)
    median_s = sorted(times)[len(times) // 2]
    print(f"  {label}: {median_s:.2f}s (median of {n_runs} runs)")
    return median_s

print("\nQuery: GROUP BY country_code WHERE date >= '2023-06-01'")
csv_time     = benchmark(csv_path, "CSV    ")
parquet_time = benchmark(parquet_path, "Parquet")
print(f"  Speedup: {csv_time / parquet_time:.1f}x")

# ── Column pruning benchmark ──────────────────────────────────────────────
# Select only 3 columns out of 56 total
PRUNING_QUERY = """
    SELECT SUM(amount), AVG(is_fraud)
    FROM '{path}'
    WHERE country_code = 'US'
"""

print("\nQuery: SELECT 2 cols WHERE country_code = 'US' (out of 56 total columns)")
csv_time     = benchmark(csv_path, "CSV    ")
parquet_time = benchmark(parquet_path, "Parquet")
print(f"  Speedup: {csv_time / parquet_time:.1f}x  (column pruning benefit)")

Delta Lake: Time Travel and ACID Writes

# pip install deltalake
from deltalake import DeltaTable, write_deltalake
import pandas as pd
import numpy as np

TABLE_PATH = "/tmp/delta_table"

# ── Version 0: Initial data write ─────────────────────────────────────────
df_v0 = pd.DataFrame({
    "user_id":   [1, 2, 3, 4, 5],
    "status":    ["free", "paid", "free", "paid", "free"],
    "spend_90d": [120.0, 890.0, 45.0, 2100.0, 300.0],
    "updated_at": pd.Timestamp("2024-01-15"),
})

write_deltalake(TABLE_PATH, df_v0, mode="overwrite")
print(f"Version 0 written: {len(df_v0)} rows")

# ── Version 1: Add new users ──────────────────────────────────────────────
df_new_users = pd.DataFrame({
    "user_id":   [6, 7, 8],
    "status":    ["free", "paid", "free"],
    "spend_90d": [0.0, 450.0, 15.0],
    "updated_at": pd.Timestamp("2024-02-01"),
})

write_deltalake(TABLE_PATH, df_new_users, mode="append")
print(f"Version 1 written (append): {len(df_new_users)} new rows")

# ── Version 2: Update existing user spend ────────────────────────────────
df_updates = pd.DataFrame({
    "user_id":   [1, 2, 3, 4, 5, 6, 7, 8],
    "status":    ["paid", "paid", "paid", "paid", "free", "free", "paid", "free"],
    "spend_90d": [500.0, 1100.0, 200.0, 2800.0, 350.0, 0.0, 600.0, 20.0],
    "updated_at": pd.Timestamp("2024-03-01"),
})

write_deltalake(TABLE_PATH, df_updates, mode="overwrite")
print(f"Version 2 written (overwrite/update): {len(df_updates)} rows")

# ── Read current version ──────────────────────────────────────────────────
dt = DeltaTable(TABLE_PATH)
current_df = dt.to_pandas()
print(f"\nCurrent version ({dt.version()}):")
print(current_df.to_string(index=False))

# ── Time travel: read Version 0 ───────────────────────────────────────────
dt_v0 = DeltaTable(TABLE_PATH, version=0)
df_at_v0 = dt_v0.to_pandas()
print(f"\nVersion 0 (initial state, January 2024):")
print(df_at_v0.to_string(index=False))

# ── Time travel: read Version 1 ───────────────────────────────────────────
dt_v1 = DeltaTable(TABLE_PATH, version=1)
df_at_v1 = dt_v1.to_pandas()
print(f"\nVersion 1 (after new users added, February 2024):")
print(df_at_v1.to_string(index=False))

# ── Inspect table history ─────────────────────────────────────────────────
history = dt.history()
print(f"\nTable history ({len(history)} versions):")
for entry in history:
    print(f"  Version {entry.get('version', '?')}: "
          f"operation={entry.get('operation', '?')}, "
          f"timestamp={entry.get('timestamp', '?')}")

# ── ML use case: generate training data at a specific historical state ────
# This is the Delta Lake equivalent of a point-in-time join on the table itself
print("\nML use case: training data as of Version 0 (prevents data leakage)")
training_features = dt_v0.to_pandas()[["user_id", "status", "spend_90d"]]
print(training_features.to_string(index=False))

---

Parquet File Internal Structure Diagram

---

YouTube Resources

Title	Channel	Why Watch
Apache Parquet Internals - File Format Deep Dive	Dremio	Best technical walkthrough of row groups, column chunks, pages, and statistics with real hex dumps
Delta Lake Tutorial for Beginners	Databricks	ACID transactions, time travel, and schema evolution in Delta Lake - official Databricks content
DuckDB: The Lightweight Analytics Database	DuckDB Team (Strange Loop)	Why DuckDB is faster than Spark for local analytical queries - vectorized execution and columnar processing
Apache Iceberg vs Delta Lake vs Apache Hudi	Seattle Data Guy	Practical comparison of the three open table formats with real use case recommendations
Columnar vs Row Storage - Why Parquet Dominates Big Data	Fireship	Fast 10-minute visual explanation of why columnar is faster for analytics workloads

---

Production Engineering Notes

The Small Files Problem

Object storage systems (S3, GCS) are optimized for large sequential reads, not many small random reads. Each file open is a separate HTTP request with overhead (latency, metadata lookup, authentication). A table with 100,000 files of 1 MB each is far slower to query than one with 100 files of 1 GB each, even though the total data is identical.

The small files problem arises naturally when pipelines write one file per partition per batch:

• Hourly pipeline × 365 days × 24 hours × 30 country partitions = 262,800 files per year
• Each 1-hour partition for a low-volume country might be 50 KB

Solutions:

1. Compaction jobs: periodically merge small files into larger ones. Delta Lake and Iceberg have built-in OPTIMIZE / rewrite_data_files() operations
2. Write fewer, larger files: increase batch size or time window before writing
3. Fewer partitions: don't partition by high-cardinality columns; partition by date (not hour) unless your data volume justifies hourly partitioning

For ML training pipelines that read many small files, the overhead can dominate actual read time. A benchmark rule: if you have more than one file per 128 MB of data, you probably have a small files problem.

Partition Pruning vs. Predicate Pushdown

These are different mechanisms that work at different layers:

Partition pruning happens at the file listing level. When a Parquet dataset is partitioned by year_month=2024-01/, querying with WHERE year_month = '2024-03' causes the query engine to list only the files in the 2024-03/ directory. Files in other partitions are never opened. This is the coarsest and most powerful optimization.

Predicate pushdown happens at the Parquet file level. Within a single Parquet file, the row group statistics are used to skip row groups that can't contain matching rows. This is finer-grained than partition pruning but less powerful.

Use both: partition by the most common filter column (usually date), and ensure your within-file data is sorted or clustered on the next most common filter column (often entity ID or category) to maximize row group skipping.

Format Choice by Pipeline Stage

Stage	Recommended Format	Reason
Raw event ingestion (Kafka)	Avro	Schema evolution, streaming-native
Raw landing zone (S3, GCS)	Parquet + Zstd	Compressed, queryable
Feature store offline	Parquet in Delta/Iceberg	Time travel, ACID, queryable
ML training data	Parquet + Zstd (partitioned)	Column pruning, predicate pushdown
Model artifacts	Custom (pickle, ONNX, TF SavedModel)	Framework-specific
Serving cache	Binary (Protocol Buffers, MessagePack)	Low-latency deserialization
Data exchange (APIs)	JSON	Universally readable, schema-explicit

S3 Byte-Range Reads

PyArrow and most Parquet libraries support byte-range reads against S3 - reading only specific byte ranges from a file without downloading the entire file. This is how column pruning and predicate pushdown work against remote object storage: the library reads the Parquet footer (last few KB), determines the byte range of the needed column chunks and row groups, and issues targeted HTTP range requests.

For large files (1 GB+), this makes remote Parquet reads nearly as efficient as local reads, because you only pay network I/O for the data you actually need.

Optimal file sizes for S3 + Parquet: 128 MB to 1 GB per file. Files smaller than 128 MB have proportionally high overhead from metadata reads and HTTP request setup. Files larger than 1 GB don't parallelize well (a single file can only be read by one thread per row group).

---

Common Mistakes

:::danger Storing Large ML Datasets as CSV or JSON

CSV and JSON are appropriate for small datasets (under 1 GB), data exchange, and human debugging. They are not appropriate as the primary storage format for ML training data at scale.

Problems with CSV for ML:

• No column pruning - reading 3 columns reads all columns
• No predicate pushdown - filtering reads the entire file
• No schema enforcement - column types can drift silently
• gzip compression is not splittable - one file, one reader
• 3-10x larger than equivalent Parquet + Zstd

If your pipeline currently uses CSV and your training jobs take more than 30 minutes, converting to Parquet is likely the single highest-leverage optimization available to you. :::

:::danger Dictionary Encoding Explosion on High-Cardinality String Columns

Storing high-cardinality string columns (UUIDs, session IDs, raw log lines) in Parquet with dictionary encoding will cause the dictionary to grow unboundedly within each row group, eventually falling back to plain encoding with significant write-time memory overhead.

Symptoms: Parquet write is very slow or runs out of memory. Parquet file is much larger than expected.

Fix: store IDs as integers (uint64) where possible. If string IDs are unavoidable, disable dictionary encoding for those specific columns:

# PyArrow: disable dictionary encoding for specific columns
from pyarrow import parquet as pq
pq.write_table(
    table,
    path,
    column_encoding={
        "user_uuid": "PLAIN",   # Disable dict encoding for UUID column
        "session_id": "PLAIN",  # Disable for high-cardinality string
    }
)

:::

:::warning Writing Many Small Parquet Files

A common pattern in PySpark or streaming pipelines: each partition writes its own Parquet file. If your data has 1000 Spark partitions and each is 2 MB, you write 2000 MB as 1000 files. Later reads open 1000 files, each with its own HTTP request to S3 - massively slower than reading 16 files of 125 MB each.

Before writing training data for ML, coalesce or repartition to control the output file count:

# PySpark: control output file count
df.coalesce(20).write.parquet(output_path)          # Reduce to 20 files
# or
df.repartition(50, "date_partition").write.parquet(output_path)  # 50 files, organized by date

Target: 128 MB to 512 MB per output file. :::

:::warning Ignoring Row Group Size When Optimizing for Parallel Reads

PyArrow's default row group size is 1 million rows or ~64 MB. For ML training with many parallel readers (16-core machine, PyTorch DataLoader with 8 workers), you want enough row groups so each reader gets its own row group to work on independently.

A 10 GB Parquet file with 1 row group can only be read by one thread at a time - the other 15 threads sit idle.

With 128 MB row groups, the same file has ~80 row groups and all 16 threads can read independently.

Tune row_group_size based on your expected number of parallel readers × file size, targeting 1-4 row groups per reader per file. :::

---

Interview Q&A

Q1: Why would you choose Parquet over CSV for ML training data, and what Parquet settings would you tune?

Why Parquet: Three compounding benefits over CSV.

First, column pruning: a typical ML training dataset has 100-500 feature columns but a model may use 20-50. Parquet reads only the requested columns; CSV reads all columns and discards the rest. At 200 columns and 50% utilization, that's 4x less I/O before any other optimization.

Second, predicate pushdown: Parquet stores min/max statistics per row group per column. A time-range filter like WHERE date >= '2024-01-01' causes the engine to skip all row groups where max(date) < '2024-01-01' without reading the data. For a 2-year history where you train on the last 3 months, predicate pushdown can skip ~85% of the file.

Third, compression: columnar layout enables type-homogeneous compression. A country_code column with 50 unique values across 100M rows is perfectly suited for dictionary encoding. Real-world Parquet + Zstd typically achieves 5-15x compression over raw CSV.

Settings to tune:

• compression="zstd" with compression_level=3 - better ratio than Snappy, faster decompress than gzip
• row_group_size: 128-512 MB for training data read by many parallel workers; larger (256 MB+) for archival
• use_dictionary=True (default) - disable only for known high-cardinality string columns
• write_page_index=True - enables column-level page statistics for finer-grained pushdown (Parquet 2.0)
• Bloom filters on high-cardinality key columns used in equality filters (user_id, merchant_id)
• Partitioning by date at the file/directory level for coarse-grained partition pruning

---

Q2: What is predicate pushdown and how does it work in Parquet?

Predicate pushdown moves filter evaluation from the compute layer (where Python, Spark, or Trino runs) down to the storage layer (where the Parquet file is read). Instead of reading all data and then discarding non-matching rows, the storage layer uses metadata to skip entire row groups that can't contain matching rows.

How it works mechanically:

1. The query planner identifies predicates in the WHERE clause
2. The Parquet reader loads the file footer - a small metadata block at the end of the file containing row group metadata, including min/max statistics and null counts for every column
3. For each row group, the reader evaluates whether the predicate could match. For a predicate amount > 1000, if max(amount) in a row group is 850, no row in that row group can possibly satisfy the predicate - the entire row group is skipped
4. Only row groups that might contain matching rows are read

Effectiveness depends on data layout: if amount values are randomly distributed across row groups, the min/max statistics overlap heavily and few row groups are skipped. If the data is sorted or clustered by amount, min/max statistics partition cleanly and most row groups can be skipped.

This is why writing training data sorted by the most common filter column (usually the timestamp or partition key) dramatically improves read performance. The ZORDER BY operation in Delta Lake achieves this for multiple columns simultaneously through z-order (space-filling curve) clustering.

---

Q3: What is the difference between Delta Lake, Iceberg, and Hudi? When would you choose each?

All three are open table formats - metadata layers on top of Parquet that add ACID transactions, time travel, and schema evolution to object storage. They differ in design philosophy and ecosystem integration.

Delta Lake: easiest to adopt if you're already on Databricks or PySpark. Best Spark integration, largest community, simplest operational model. The transaction log is a sequence of JSON files in _delta_log/ - easy to inspect and debug manually. Choose Delta if your team uses Spark heavily and wants the simplest path to production.

Apache Iceberg: the most technically rigorous design. Better support for multiple query engines (Spark, Flink, Trino, Dremio) without vendor lock-in. Stronger support for partition evolution (change partitioning scheme without rewriting data). Better performance for tables with very large numbers of partitions (millions). Choose Iceberg if you're building a multi-engine lakehouse or need to avoid Databricks dependency.

Apache Hudi: built at Uber for incremental data processing. Best support for upserts and CDC (Change Data Capture) workflows. Two storage types: Copy-on-Write (better read performance) and Merge-on-Read (better write performance, with merge at read time). Choose Hudi if you're applying database change streams (CDC) to an analytical table or need upsert performance.

For ML specifically, all three provide the key value: time travel for point-in-time training data. You can query the feature table as it existed at any past timestamp, which is the equivalent of SCD Type 2 querying without the SCD overhead.

---

Q4: How would you benchmark storage format performance for a specific ML pipeline?

Benchmarking storage formats is workload-specific - the right format depends on your actual access patterns, not synthetic benchmarks. A structured benchmark:

1. Define representative queries: identify the 3-5 most common queries your pipeline runs - typically a date-range filter + specific column selection + possible group-by
2. Generate realistic data: use production data or a schema-faithful synthetic dataset of the same size and cardinality characteristics
3. Write each format: CSV (gzip), Parquet (Snappy), Parquet (Zstd), Delta Lake, ORC
4. Measure cold reads: clear OS page cache between reads (echo 3 > /proc/sys/vm/drop_caches on Linux) to measure actual I/O performance, not cache effects
5. Measure warm reads: run the same query twice and report both - the second read often benefits from OS page cache and represents repeated training runs
6. Track three metrics: wall-clock time, data scanned (actual bytes read), storage size

Tools: DuckDB is the most convenient for local benchmarking - it supports CSV, Parquet, and Delta Lake natively, shows bytes scanned in EXPLAIN ANALYZE, and runs without cluster setup. For distributed benchmarks, Spark with detailed event logs and the Spark History Server.

Key insight: don't just benchmark scan speed - include write time and storage cost. A format that's 20% faster to read but 50% slower to write may not win if your pipeline writes often.

---

Q5: What is the small files problem and how do you fix it for an ML training pipeline?

The small files problem: when a dataset consists of thousands of small files instead of tens of large files, query performance degrades severely even if the total data volume is the same. Each file requires:

• An HTTP request to S3 (typically 50-200ms latency each)
• Parquet footer read (another HTTP request)
• Query planning overhead per file

With 10,000 files of 1 MB each, you might spend more time opening files than reading data. With 100 files of 100 MB each, the same 10 GB dataset reads nearly at network saturation speed.

Causes in ML pipelines:

• Partitioning by high-cardinality columns (user country × date × device type = many small partitions for low-volume combinations)
• Streaming pipelines that write one file per micro-batch per partition
• Incremental append pipelines that add small files without compaction

Fixes:

1. Compaction at write time: coalesce before writing - df.coalesce(50).write.parquet(path) in PySpark
2. Scheduled compaction: for Delta/Iceberg tables, run OPTIMIZE weekly to merge small files into target-sized files (Delta) or rewrite_data_files() (Iceberg)
3. Partition strategy review: avoid partitioning by columns with more than 100-200 distinct values; partition by date only, not by date + hour + country
4. File size targets: aim for 128 MB - 1 GB per file for analytical workloads; enforce with compaction jobs if streaming writes produce smaller files

For an ML training pipeline specifically: run a weekly compaction job that merges all files into the target size range, then generate the weekly training dataset from the compacted table. Training reads should never touch uncompacted small files.

---

:::tip Format Decision Cheatsheet

• CSV: small datasets, data exchange, debugging, human inspection
• JSON: API responses, config files, semi-structured data with variable schema
• Parquet + Zstd: ML training data, analytical queries, feature store offline storage - use this by default
• Parquet + Snappy: when decompression speed matters more than ratio (GPU training bottlenecked on CPU preprocessing)
• Avro: Kafka event streaming, schema-evolution-critical pipelines, cross-language serialization
• ORC: Hive-native workloads, existing Hive infrastructure
• Delta Lake: Spark-based lakehouses, when you need ACID + time travel + Databricks ecosystem
• Iceberg: multi-engine lakehouses, large partition counts, avoiding vendor lock-in
• Hudi: CDC/upsert-heavy pipelines, Uber-style incremental processing :::

---

How to Read a Parquet File's Internals

Understanding what's inside a Parquet file helps you tune it effectively. PyArrow exposes the full metadata:

import pyarrow.parquet as pq

# Inspect file metadata without reading data
pf = pq.ParquetFile("/tmp/training_data.parquet")
meta = pf.metadata
schema = pf.schema_arrow

print(f"Format version:   {meta.format_version}")
print(f"Created by:       {meta.created_by}")
print(f"Num row groups:   {meta.num_row_groups}")
print(f"Total rows:       {meta.num_rows:,}")
print(f"Num columns:      {meta.num_columns}")
print()

# Inspect each row group
for i in range(meta.num_row_groups):
    rg = meta.row_group(i)
    print(f"Row group {i}:")
    print(f"  Rows:        {rg.num_rows:,}")
    print(f"  Total size:  {rg.total_byte_size / 1e6:.2f} MB")

    # Inspect specific column chunk within the row group
    for col_idx in range(min(3, meta.num_columns)):
        col = rg.column(col_idx)
        stats = col.statistics
        print(f"  Column '{schema.field(col_idx).name}':")
        print(f"    Encoding:    {col.encodings}")
        print(f"    Compression: {col.compression}")
        if stats and stats.has_min_max:
            print(f"    Min: {stats.min}")
            print(f"    Max: {stats.max}")
        print(f"    Null count:  {stats.null_count if stats else 'N/A'}")
    print()

# Read schema
print("Schema:")
print(schema.to_string())

This is the exact metadata the query engine reads from the footer before deciding which row groups to scan. If you run this on a training dataset, you'll see whether min/max statistics are well-separated (good predicate pushdown potential) or heavily overlapping (poor pushdown).

---

Key Takeaways

• Columnar formats win for ML training because ML queries read a few columns across many rows - the opposite of OLTP access patterns. Column pruning and predicate pushdown are first-order optimizations.
• Parquet + Zstd is the default for most ML pipelines: best compression-to-decompression-speed ratio, wide ecosystem support, and native integration with every major compute engine.
• Row group size controls parallelism: tune to 128-256 MB for read-heavy workloads; ensure enough row groups for all parallel readers to work concurrently.
• Dictionary encoding is automatic but breaks on high-cardinality strings: store IDs as integers where possible; disable dictionary encoding explicitly for UUID/session ID columns.
• Predicate pushdown requires sorted or clustered data: write data sorted by the most common filter column (usually timestamp) to maximize row group skipping.
• Avro belongs in Kafka: it's schema-evolution-first and row-oriented. Great for streaming serialization; wrong for analytical reads.
• Open table formats (Delta, Iceberg, Hudi) add the database features that Parquet lacks: ACID transactions, time travel, schema evolution, and compaction on object storage. Use them for feature stores and production data lakes.
• The small files problem is the most common cause of unexpectedly slow Parquet reads: target 128 MB - 1 GB per file; use compaction (OPTIMIZE in Delta, rewrite_data_files in Iceberg) to fix accumulation of small files.
• Compression codec choice is workload-dependent: Zstd for I/O-bound workloads (most ML training); Snappy/LZ4 for CPU-bound workloads where decompression speed dominates.