Data Engineering # Column-Oriented Storage Formats: Parquet, Lance, Delta Lake, and Iceberg Architecture and Optimization 2025-12-129 min read Lucio Durán Engineering Manager & AI Solutions Architect Also available in: [Español]() ## Why Columns Beat Rows for Analytics Consider a table with 100 columns and 1 billion rows. Your query needs 3 of those columns. Row-oriented storage (PostgreSQL, MySQL): ``` Row 1: [col1, col2, col3, ..., col100] ← reads 100 columns to get 3 Row 2: [col1, col2, col3, ..., col100] ← reads 100 columns to get 3 ... Row 1B: [col1, col2, col3, ..., col100] ``` To read 3 columns, you read all 100 for every row. If each column is 8 bytes, that's 800GB of I/O for 24GB of useful data. Column-oriented storage (Parquet): ``` Column chunk 1: [row1_val, row2_val, ..., row1B_val] ← skip Column chunk 2: [row1_val, row2_val, ..., row1B_val] ← skip Column chunk 3: [row1_val, row2_val, ..., row1B_val] ← READ ... Column chunk 47: [row1_val, row2_val, ..., row1B_val] ← READ ... Column chunk 92: [row1_val, row2_val, ..., row1B_val] ← READ ``` You read exactly the 3 column chunks you need: 24GB. That's a 33x reduction in I/O, and on S3 where you pay per byte transferred, it's also a 33x reduction in cost. ## Apache Parquet Internals [Apache Parquet]() is the de facto standard columnar format. Understanding its internal structure is essential for optimization. ### File Layout ``` ┌──────────────────────────────────┐ │ Magic: PAR1 │ ├──────────────────────────────────┤ │ Row Group 0 │ │ ├─ Column Chunk: user_id │ │ │ ├─ Page 0 (data page) │ │ │ ├─ Page 1 (data page) │ │ │ └─ Page 2 (data page) │ │ ├─ Column Chunk: timestamp │ │ │ ├─ Page 0 │ │ │ └─ Page 1 │ │ └─ Column Chunk: amount │ │ └─ Page 0 │ ├──────────────────────────────────┤ │ Row Group 1 │ │ ├─ Column Chunk: user_id │ │ ├─ Column Chunk: timestamp │ │ └─ Column Chunk: amount │ ├──────────────────────────────────┤ │ Footer (metadata) │ │ ├─ Schema │ │ ├─ Row group metadata │ │ │ ├─ Column chunk offsets │ │ │ ├─ Min/Max statistics │ │ │ ├─ Null counts │ │ │ └─ Distinct counts │ │ └─ Key-value metadata │ ├──────────────────────────────────┤ │ Footer length (4 bytes) │ │ Magic: PAR1 │ └──────────────────────────────────┘ ``` **Row Groups** are the unit of parallelism. Each row group contains a horizontal slice of rows (typically 128MB worth). Query engines process row groups in parallel across cores. **Column Chunks** are the unit of I/O. Each column chunk contains all values for one column within one row group. When you ` SELECT amount `, the engine reads only the ` amount ` column chunks, skipping everything else. **Pages** are the unit of encoding and compression. Each page (typically 1MB) is independently compressed and encoded. This means decompression can be parallelized within a column chunk. ### Predicate Pushdown: The Real Performance Win The footer contains min/max statistics for each column chunk. When you write ` WHERE timestamp > '2025-01-01' `, the query engine checks each row group's timestamp min/max. If a row group's max timestamp is ` 2024-06-15 `, the entire row group is skipped, no decompression, no I/O for that data. ```python # This is what happens internally during predicate pushdown def should_read_row_group(row_group_metadata, predicate): col_stats = row_group_metadata.column('timestamp').statistics if predicate.op == '>': # If max value is less than predicate, skip entire row group if col_stats.max < predicate.value: return False # Skip ~1M rows with zero I/O return True ``` But here's the critical catch: **predicate pushdown only works if your data is sorted (or at least clustered) by the filtered column.** If your data is randomly ordered, every row group contains the full range of timestamps, and no row groups get skipped. ```python # Bad: random order → every row group has min=2020, max=2026 # Predicate pushdown skips 0 row groups df_random = spark.read.parquet("s3://data/random-order/") df_random.filter("timestamp > '2025-01-01'").count() # Reads: 12TB (all row groups) # Good: sorted by timestamp → row groups have tight min/max ranges # Predicate pushdown skips old row groups df_sorted = spark.read.parquet("s3://data/sorted-by-time/") df_sorted.filter("timestamp > '2025-01-01'").count() # Reads: 1.2TB (only recent row groups) ``` That's a 10x difference from just sorting before writing. The importance of this cannot be overstated. ### Encoding Schemes Parquet supports multiple encoding schemes that compress data before general-purpose compression (ZSTD, Snappy) is applied: ``` Dictionary Encoding: Original: ["USD", "EUR", "USD", "USD", "GBP", "EUR", "USD"] Dictionary: {0: "USD", 1: "EUR", 2: "GBP"} Encoded: [0, 1, 0, 0, 2, 1, 0] → Strings reduced to integers. Huge savings for low-cardinality columns. Run-Length Encoding (RLE): Original: [1, 1, 1, 1, 1, 2, 2, 2, 3, 3] Encoded: [(1, 5), (2, 3), (3, 2)] → Consecutive repeated values compressed to (value, count). Delta Encoding: Original: [1000, 1002, 1005, 1007, 1010] Encoded: [1000, 2, 3, 2, 3] → Store differences. Great for timestamps and monotonically increasing IDs. ``` ## Delta Lake and Iceberg: Tables on Top of Parquet Raw Parquet files are immutable, you can't update a row or run a transaction. Delta Lake and Iceberg add a metadata layer that provides: - **ACID transactions**: Multiple writers can safely update data concurrently - **Time travel**: Query data as it existed at any point in time - **Schema evolution**: Add, rename, or widen columns without rewriting data - **Partition evolution**: Change partitioning strategy without rewriting ### Delta Lake Transaction Log Delta Lake stores a transaction log in ` _delta_log/ ` alongside the Parquet files: ``` my_table/ ├── _delta_log/ │ ├── 00000000000000000000.json ← initial commit │ ├── 00000000000000000001.json ← add 3 files │ ├── 00000000000000000002.json ← update: remove 1, add 1 │ └── 00000000000000000003.json ← delete: remove 2 ├── part-00000-abc123.parquet ├── part-00001-def456.parquet └── part-00002-ghi789.parquet ``` Each JSON file records which Parquet files were added or removed: ```json {"add": {"path": "part-00000-abc123.parquet", "size": 134217728, "partitionValues": {"date": "2025-01-15"}, "stats": "{\"numRecords\":1048576,\"minValues\":{\"amount\":0.01},\"maxValues\":{\"amount\":9999.99}}"}} ``` ### Apache Iceberg Metadata Iceberg takes a different approach with a tree-structured metadata hierarchy: ``` Catalog → Table Metadata File → Manifest List → Manifest Files → Data Files Table Metadata (metadata/v3.metadata.json): Current snapshot: snap-12345 Schema: {id: long, name: string, amount: decimal} Partition spec: date(timestamp) Manifest List (snap-12345-m0.avro): Manifest 1: manifest-abc.avro (date=2025-01-*, 500 files) Manifest 2: manifest-def.avro (date=2025-02-*, 480 files) Manifest (manifest-abc.avro): file1.parquet: rows=1M, min_amount=0.01, max_amount=500.00 file2.parquet: rows=1M, min_amount=0.50, max_amount=9999.99 ``` This tree structure means Iceberg can prune at three levels before reading any data: skip entire manifests (partition pruning), skip files within manifests (min/max pruning), then skip row groups within files (Parquet-level pushdown). ## Lance: Column Storage for ML Lance is the newer format I've been working with for ML pipelines. It's column-oriented like Parquet but optimized for different access patterns: ```python import lance # Write a dataset with embeddings ds = lance.write_dataset( pa.table({ "id": range(1_000_000), "text": texts, "embedding": embeddings, # 768-dim float32 vectors "metadata": metadata_dicts, }), "s3://ml-data/documents.lance", ) # Fast vector search, this is where Lance shines results = ds.to_table( nearest={"column": "embedding", "q": query_vector, "k": 10}, filter="metadata.category = 'technical'" ) ``` Lance stores vectors in a format optimized for approximate nearest neighbor (ANN) search, while keeping tabular data columnar. The key architectural difference from Parquet: - **Random access by row ID**: Lance supports O(1) row lookup, while Parquet requires scanning to find a specific row - **Append and update**: Lance supports appending new data and updating existing rows without rewriting entire files - **Vector index**: Built-in IVF-PQ index for similarity search ``` Lance File Layout: ┌──────────────────────┐ │ Data Section │ │ ├─ Batch 0 │ │ │ ├─ Column: id │ │ │ ├─ Column: text │ │ │ └─ Column: embed │ │ └─ Batch 1 │ ├──────────────────────┤ │ Index Section │ │ ├─ IVF centroids │ │ └─ PQ codebooks │ ├──────────────────────┤ │ Manifest │ │ ├─ Schema │ │ ├─ Fragment metadata│ │ └─ Version info │ └──────────────────────┘ ``` ## Practical Optimization: Real Queries Here's a real optimization I did on a 4TB clickstream dataset: ```sql -- Before optimization: 47 seconds, scans 4TB SELECT page_category, COUNT(*) as views, COUNT(DISTINCT user_id) as unique_users, AVG(time_on_page_seconds) as avg_time FROM clickstream WHERE event_date BETWEEN '2025-01-01' AND '2025-01-31' GROUP BY page_category; ``` Optimizations applied: 1. **Partition by date**: Reduced scan from 4TB to 340GB (January only) 2. **Sort within partitions by page\_category**: Predicate pushdown on GROUP BY column 3. **Dictionary encode page\_category**: 50 unique values → integers 4. **ZSTD level 3**: 20% smaller files than Snappy default ```python # Writing optimized Parquet with PyArrow import pyarrow.parquet as pq writer = pq.ParquetWriter( 'clickstream.parquet', schema=schema, compression='zstd', compression_level=3, use_dictionary=['page_category', 'browser', 'country'], row_group_size=64 * 1024 * 1024, # 64MB row groups for this workload write_statistics=True, ) # Sort before writing, this is critical df_sorted = df.sort_values(['event_date', 'page_category']) writer.write_table(pa.Table.from_pandas(df_sorted)) writer.close() ``` After optimization: 2.1 seconds. Same data, same query, 22x faster. The breakdown: - Partitioning saved 91.5% of I/O (only read January) - Predicate pushdown on sorted data saved another 60% within January - Dictionary encoding reduced page\_category column from 340MB to 12MB - ZSTD compression saved 20% on remaining data ## Choosing Between Formats My decision tree after working with all four: **Pure Parquet** when you have read-only analytics data, no concurrent writers, and want maximum compatibility. Every engine reads Parquet. **Delta Lake** when you need ACID transactions and you're in the Databricks/Spark ecosystem. Delta's ` OPTIMIZE ` and ` Z-ORDER ` commands are excellent for hands-off compaction and multi-dimensional clustering. **Apache Iceberg** when you have multiple query engines (Trino + Spark + Flink) and need partition evolution. Iceberg's hidden partitioning is genuinely better than Delta's explicit partitioning for evolving schemas. **Lance** when your data includes vector embeddings and your access pattern mixes analytics queries with similarity search. Don't use Lance for pure tabular analytics, Parquet is more mature and better supported. The reality is that for most teams, the choice between Delta and Iceberg matters less than proper data sorting, compression tuning, and partition strategy. Get those fundamentals right on either format and you'll see 10-50x improvements over naive usage. I've watched teams agonize over Delta vs. Iceberg for months while their queries scan terabytes of unsorted, unpartitioned Parquet. Fix the data layout first. The table format is a footnote. ## Frequently asked questions ### Why is columnar storage faster for analytics than row-oriented storage? Analytics queries typically read a few columns across millions of rows (e.g., SELECT AVG(price) WHERE date \> X). Columnar storage puts all values of a column contiguously on disk, so reading one column does sequential I/O on just that data. Row storage forces you to read entire rows including columns you don't need, wasting bandwidth and cache. ### What's the difference between Delta Lake and Apache Iceberg? Both are table formats that add ACID transactions, time travel, and schema evolution on top of Parquet files. Delta Lake uses a JSON-based transaction log, was created by Databricks, and has the deepest Spark integration. Iceberg uses Avro-based metadata with a catalog abstraction, was created at Netflix, and has broader multi-engine support (Trino, Flink, Spark, Athena). ### When should I use Lance instead of Parquet? Lance is optimized for ML workloads with high-dimensional vector data and random access patterns. If you need fast vector similarity search alongside tabular data, or your access pattern is random row lookups rather than columnar scans, Lance outperforms Parquet significantly. For pure analytics queries, Parquet is still the standard. ## Tips 1. **Tip:** Set your Parquet row group size to match your query patterns. For point lookups, smaller row groups (8-16MB) reduce I/O. For full scans, larger row groups (128-256MB) reduce metadata overhead. The default 128MB is a good starting point, but 4x speedups from tuning this on specific workloads are common. 2. **Tip:** Always sort your data by the most-filtered column before writing to Parquet. This makes min/max statistics in the row group metadata actually useful for predicate pushdown. Unsorted data means every row group's min/max spans the entire value range, and the query engine reads everything anyway. 3. **Tip:** Use ` ZSTD ` compression level 3 for Parquet in production, not the default level 1. Level 3 gives 15-20% better compression with negligible encoding overhead. Level 19 gives diminishing returns and kills write throughput. Benchmarking all 22 levels confirms that 3 is the sweet spot. 4. **Tip:** When choosing between Delta Lake and Iceberg, check your existing ecosystem. If you're on Databricks, Delta is native. If you're on AWS Athena/Glue or Snowflake, Iceberg has better integration. The formats are converging, don't pick based on features, pick based on what your query engine supports natively. - parquet - lance - delta-lake - iceberg - column-storage - data-engineering - analytics - query-optimization [Back to Blog]()