Columnar Formats and the Lakehouse

"They asked for two columns. I, a humble row, handed over all twenty and apologized for the wait."

A Record That Reads Everything to Answer Anything

A training pipeline and an analytics pipeline ask the same physical question of storage thousands of times a day: read these few attributes from these many records. A churn-feature job wants two columns out of forty; a label-distribution check wants one column out of two hundred. How the bytes are arranged on the object store decides whether that query reads what it asked for or drags the entire dataset off disk to discard most of it. Section 8.2 settled where the data lives (object storage, distributed filesystems). This section settles how it is encoded there, and that single choice moves analytical and feature-read cost by an order of magnitude.

1. Row Layout Versus Columnar Layout Beginner

A table is a logical grid, but disk is one-dimensional, so the grid must be flattened into a byte sequence, and there are two natural ways to do it. A row store writes record 1 in full (all its columns), then record 2 in full, and so on; the fields of one row sit physically adjacent. This is ideal when you fetch whole records by key, which is why transactional databases default to it. A columnar store does the opposite: it writes all values of column 1 together, then all values of column 2, and so on; the values of one column sit physically adjacent. Parquet and ORC are the dominant columnar formats on object storage, and the layout difference is the entire story of why analytical scans are cheap on them.

Two consequences follow directly from putting a column's values next to each other. The first is column pruning: a query that needs two of twenty columns reads only those two contiguous regions and skips the rest, so bytes scanned fall in proportion to columns touched, not rows kept. The second is compression: values within one column share a type and a distribution, and once similar values are adjacent, dictionary encoding, run-length encoding, and bit packing all become dramatically effective, far more than they ever are on the mixed-type byte soup of a row. Figure 8.3.1 shows both layouts side by side and marks the bytes a two-column query must read in each.

2. Measuring the Two Wins: Pruning and Compression Intermediate

The two benefits are easy to state and easy to measure, so we measure them. The code below builds a 200,000-row table of twenty columns: nineteen noisy float columns plus one low-cardinality categorical column (a region code with five distinct values), arranged in clustered runs the way a real table sorted on a key would be. It then computes, from an explicit byte model, how much a row store must read for a full scan versus how much a columnar store reads when a query asks for just two of the twenty columns. It measures the columnar compression ratio on the clustered categorical column. Finally, when pyarrow is installed, it writes a real Parquet file and reads back the compressed size of only the two requested column chunks, so the from-scratch model can be checked against a production encoder.

import io, random

random.seed(7)
N_ROWS, N_COLS, READ_COLS = 200_000, 20, 2     # a query touching 2 of 20 columns

cols = {f"f{c}": [random.gauss(0, 1) for _ in range(N_ROWS)] for c in range(N_COLS - 1)}
categories = ["US", "EU", "APAC", "LATAM", "MEA"]
region = []                                     # low-cardinality key in clustered runs
while len(region) < N_ROWS:
    region.extend([categories[random.randint(0, 4)]] * random.randint(200, 2000))
cols["region"] = region[:N_ROWS]

BF, BC = 8, 8                                   # bytes per float, per category slot
row_bytes = (N_COLS - 1) * BF + BC              # a full row record
row_store_full_scan = row_bytes * N_ROWS       # row store: a query must read every row

want = ["f0", "region"]                         # columnar store reads only these two
columnar_pruned_scan = BF * N_ROWS + BC * N_ROWS
print("row-store full read (MB)   :", f"{row_store_full_scan / 1e6:.1f}")
print("columnar pruned read (MB)  :", f"{columnar_pruned_scan / 1e6:.1f}")
print("bytes-scanned reduction    :", f"{row_store_full_scan / columnar_pruned_scan:.1f}x")

# Compression: similar values adjacent => dictionary + run-length collapses the column.
raw = BC * N_ROWS
codes = [categories.index(v) for v in cols["region"]]
runs = 1 + sum(codes[i] != codes[i - 1] for i in range(1, len(codes)))
rle = sum(len(s) for s in categories) + 5 + runs * 5   # tiny dict + (code, run-length)
print("region raw (MB)            :", f"{raw / 1e6:.2f}")
print("region dict+RLE (MB)       :", f"{rle / 1e6:.2f}")
print("columnar compression ratio :", f"{raw / rle:.1f}x")

try:                                            # cross-check against a real Parquet writer
    import pyarrow as pa, pyarrow.parquet as pq
    table = pa.table(cols); buf = io.BytesIO()
    pq.write_table(table, buf, compression="snappy")
    md = pq.ParquetFile(io.BytesIO(buf.getvalue())).metadata
    idx = {table.schema.field(i).name: i for i in range(table.num_columns)}
    pruned = sum(md.row_group(g).column(idx[n]).total_compressed_size
                 for g in range(md.num_row_groups) for n in want)
    print("parquet total (MB)         :", f"{buf.getbuffer().nbytes / 1e6:.2f}")
    print("parquet 2-col chunks (MB)  :", f"{pruned / 1e6:.2f}")
    print("parquet column-prune ratio :", f"{buf.getbuffer().nbytes / pruned:.1f}x")
except Exception as e:
    print("pyarrow not available:", type(e).__name__)

row-store full read (MB)   : 32.0
columnar pruned read (MB)  : 3.2
bytes-scanned reduction    : 10.0x
region raw (MB)            : 1.60
region dict+RLE (MB)       : 0.00
columnar compression ratio : 2075.2x
parquet total (MB)         : 35.57
parquet 2-col chunks (MB)  : 1.87
parquet column-prune ratio : 19.0x

The numbers separate the two effects cleanly. Column pruning alone gives the $10\times$ scan reduction from the byte model: two columns instead of twenty, with no compression assumed. The real Parquet file reaches $19\times$ because pruning and per-column compression stack, and the clustered categorical column shows why compression is a columnar privilege, collapsing by more than $2000\times$ once its repeated values lie next to one another. A row store cannot reach that ratio on the same column, because in a row layout each region value is wedged between nineteen unrelated floats and never forms a run. Real workloads sit between these extremes, but the direction is invariant: the wider the table and the more selective the query, the more a columnar layout wins.

Column pruning is not a trick the storage layer pulls off alone; the query engine has to know which columns the query actually needs and push that knowledge down to the reader. This is exactly the projection pushdown that Spark's Catalyst optimizer performs: it inspects the logical plan, determines the minimal set of columns required, and instructs the Parquet reader to skip the rest before a single irrelevant byte leaves the object store. The columnar format and the optimizer are partners; we built the optimizer side of this in Section 7.3, and Parquet is the storage side that makes its pruning pay off.

import pyarrow.parquet as pq

# Read ONLY two columns from a wide Parquet file. The reader fetches just those
# column chunks; the other eighteen columns are never decoded or transferred.
table = pq.read_table("features.parquet", columns=["f0", "region"])

# Push a predicate down too: rows are skipped using per-row-group min/max stats,
# so whole chunks are never read when they cannot match the filter.
table = pq.read_table("features.parquet", columns=["f0", "region"],
                      filters=[("region", "=", "EU")])

3. Apache Arrow: The Columnar Layout, In Memory Intermediate

Parquet is a columnar layout on disk, optimized for compact storage and fast scans. Apache Arrow is the same columnar idea in memory, optimized for fast processing and, above all, for sharing. Arrow defines a standard in-memory layout for columnar data: a column is a contiguous buffer of values plus a validity bitmap, with a byte specification precise enough that any system implementing it lays the bytes out identically. That shared specification is what makes Arrow valuable. When two systems both speak Arrow, one can hand a table to the other by passing a pointer to the buffers, with no serialization, no copy, and no per-row translation. This is the zero-copy exchange that lets a Parquet reader, a query engine, a Python dataframe, and a GPU library move data between them at memory bandwidth rather than parsing speed.

For AI pipelines the consequence is direct. A feature-engineering step in Spark can produce Arrow batches that a training data loader consumes without a serialize-then-deserialize round trip, and a vectorized user-defined function can run Python code over Arrow batches at near-native speed because the data arrives in the layout NumPy and pandas already use. That vectorized-UDF path, where Spark ships Arrow batches to Python workers instead of pickling row by row, is exactly the mechanism we examined in Section 7.8; Arrow is the wire format and the in-memory format that makes it fast. The throughline is that columnar is not only a disk decision: it is the shape data should keep as it crosses every boundary in the pipeline.

4. The Lakehouse: A Transactional Table Over Object Storage Intermediate

Columnar files on object storage are fast to scan but, by themselves, a loose pile of immutable blobs. They have no notion of a table version, no atomic way to add or replace rows, and no protection against a reader seeing a half-written update. The data lake gave us cheap, scalable storage and surrendered the guarantees a warehouse always had. The lakehouse closes that gap: it is a transactional table layer placed over the same columnar files on the same object storage, adding the warehouse's guarantees without giving up the lake's cost and scale. Delta Lake, Apache Iceberg, and Apache Hudi are the three dominant implementations, and although they differ in detail they share one mechanism: a metadata log of commits that defines, at every point in time, exactly which data files constitute the table.

That commit log is what turns a directory of Parquet files into a table with real semantics. ACID transactions mean a writer's changes appear all at once or not at all, so a reader never sees a partial commit even while a write is in flight. Schema evolution means columns can be added, renamed, or retyped through metadata, with old files reinterpreted under the new schema rather than rewritten. Time travel means every commit is a queryable version, so you can read the table exactly as it stood at a past commit or timestamp. Efficient upserts (merge, update, delete) mean you can change a few rows without rewriting the dataset, by writing new files and recording in the log which old files they supersede. Figure 8.3.2 shows the layering: immutable columnar files at the bottom, a metadata commit log defining table versions in the middle, and the engines that read and write through that log on top.

The mechanism behind every one of these features is the same idea this chapter keeps returning to. The data files are immutable shards on object storage; the table is the set of shards a given commit points to. An upsert does not edit a file in place, an operation object stores do not support well anyway; it writes new shards and atomically commits a new version of the file set. Time travel is just reading the file set named by an older commit. This is partitioning and sharding wearing a transactional hat, and it is the same arc that began as data shards in this chapter and returns as sharded parameters and model shards later in the book.

5. Why This Matters for AI: Reproducible, Versioned Training Tables Advanced

A model is only as reproducible as the data it was trained on, and "the data" on a plain data lake is a moving target: files get appended, corrected, and backfilled, so the dataset you trained on last month may no longer exist in the same form. The lakehouse fixes this by making the training set an explicit, immutable version. You pin training to a specific table commit, record that version id alongside the model checkpoint, and months later you can read the table exactly as it stood, re-run the pipeline, and get a dataset identical to the original byte for byte. Time travel turns "which data produced this model?" from an archaeology problem into a single version number, and it does so on cheap object storage rather than an expensive warehouse. This data-versioning story is foundational enough that Section 8.9 develops lineage and reproducible dataset versioning in full.

The efficient-upsert capability matters just as much for the daily reality of feature pipelines. Training and feature tables are rarely static; labels get corrected, late-arriving events backfill yesterday's partition, and a feature definition changes and must be recomputed for a subset of entities. On raw Parquet, any of these forces a full rewrite of the affected partitions. On a lakehouse table, a MERGE updates only the rows that changed by writing new files and retiring superseded ones, so a correction touches kilobytes of metadata and a handful of files rather than the whole dataset. The same transactional table that ingests a stream of corrections is also the one a training job reads from at a pinned version, which is why the lakehouse sits at the seam between batch and streaming ingestion that Chapter 9 takes up.

Columnar encoding decides what a scan costs, Arrow decides what crossing a system boundary costs, and the lakehouse decides what a table version costs, and together they turn cheap object storage into a reproducible, queryable, transactional foundation for training and feature data. What none of them decides yet is how the files are arranged within a table: how rows are grouped into partitions, how small files are compacted into large ones, and how data is sorted so that pruning and predicate pushdown actually skip the bytes they could. That layout question is where the costs in this section are won or lost in practice, and it is the subject of Section 8.4.