Data Formats: Parquet, ORC, Avro, and Arrow

Rows, columns, and the work a reader avoids

A storage object contains bytes that a query engine must interpret. To total order amounts by country, a reader needs country and amount for many rows. In a CSV, fields from the same row are adjacent; locating selected fields usually requires scanning through intervening text. In a columnar file, values from the same column are grouped, so unrelated columns can be omitted from the read. CSV can still be compressed, parsed with an explicit schema, or stopped early for a limited query. It lacks Parquet’s built-in column-chunk directory and block statistics.

Parquet and ORC are columnar storage formats. Avro serializes records and is useful for events and sequential record processing. Arrow defines a columnar memory layout and also IPC stream and file formats. They belong in the same discussion because one pipeline can receive Avro records, write Parquet files, and process decoded columns as Arrow arrays. Choosing among them starts with the reader’s work: whole records, selected columns across many rows, or exchanging existing arrays.

Inside a Parquet file

A row group is a horizontal slice of the table. Within it, each primitive leaf column has a column chunk divided into pages. The footer describes the schema and chunk locations; chunk metadata can contain statistics. Minimum, maximum, and null counts are not guaranteed to be present for every column. A reader first obtains metadata, then locates relevant chunks. Fetching the footer still requires I/O, even when all data pages can be skipped. Parquet file layout

Encoding changes the representation of values before a compression codec reduces bytes. A dictionary stores distinct values and encodes references to them; repeated countries can use small indices. Run-length and bit-packed encodings represent repetitions and small integers, while delta encodings exploit differences between adjacent values. A writer may fall back from dictionary encoding when the dictionary grows. Seeing PLAIN alongside RLE_DICTIONARY does not by itself prove fallback: dictionary values also use PLAIN. Snappy and Zstandard then operate on encoded page data. Their benefits and CPU costs depend on values, ordering, settings, and implementation; no fixed compression ranking follows from the names. PyArrow writer options

Nested structures are represented through leaf columns plus definition and repetition levels. Definition levels distinguish absent optional ancestors or values; repetition levels preserve repeated-field boundaries. The levels themselves are encoded, rather than stored as two ordinary integers beside every value. A null list, an empty list, and a list containing a null element must survive a round trip as different values. Typed nested fields let a reader select supported subfields without parsing a JSON string. Unknown or original payloads can still be retained as text when replay or schema variability requires it. Nested encoding

PAR1
Row group 0: [order_id pages] [country pages] [amount pages] ...
Row group 1: [order_id pages] [country pages] [amount pages] ...
File metadata: schema, chunk offsets, optional statistics
Footer length | PAR1

What pruning proves

Partition pruning excludes files using partition values. Projection limits required columns, including any columns needed to evaluate a filter. Predicate pushdown passes filtering work to the data reader; statistics can rule out row groups, and supported indexes or bloom filters can narrow work further. These are complementary mechanisms, not a universal three-step execution order. Metadata overlap only identifies candidates. The reader still evaluates rows in candidate groups to return exact matches. Missing statistics require a conservative decision to retain a group. Parquet reading and filtering

For the half-open interval [20 January, 21 January), a group is impossible when its maximum is before the 20th or its minimum is on or after the 21st. A group spanning both dates remains a candidate even if no row actually matches. Sorting timestamps makes group ranges narrower: the lab’s million rows span about 81 days, and its 100,000-row groups span about 8.1 days. This particular query overlaps one of ten sorted groups; a query crossing a group boundary can overlap two. After shuffling, all ten are candidates. That does not imply reading every column or every byte of the file.

The compressed chunk sizes in exercise 1 are metadata-based logical sizes, not measured storage traffic. Footer reads, range coalescing, readahead, cache hits, and page-level decisions affect actual requests and bytes. Use query plans and storage or filesystem instrumentation to measure those. Sorting also costs compute and delays writes; choose it from recurring filter patterns rather than assuming any sort always pays.

Row groups, files, and schema changes

Larger groups can improve compression and reduce metadata overhead, but require more buffering and offer coarser skipping. File size separately affects task scheduling, listing, opens, and compaction. A 128 MB row-group target and files of a few hundred MB can be starting hypotheses for a workload, not universal defaults. PyArrow’s row_group_size is a row count, whereas ORC stripe_size is expressed in bytes. Measure compressed sizes, memory, scan selectivity, and latency with the actual engine before setting a production target.

Each Parquet file has its own schema. The lab explicitly unifies schemas by name, allowing a new nullable channel column to be returned as null for older files. Reading both files under only the older schema instead returns all rows but hides channel. A rename becomes two separate names under this policy. A catalog or table format using stable field IDs can provide different rename semantics.

Type evolution also depends on the reader policy. PyArrow’s default unification rejects int64 versus string; permissive promotion can combine int32 and int64 into int64. Accepting a type is not proof that its business meaning is compatible: cents and dollars still need an explicit conversion. Define nullability, units, timestamp timezone and precision, field identity, and allowed migrations in the table contract. New nullable fields, deliberate backfills, and tested type widening are decisions rather than automatic permission to change any file. Schema unification rules

ORC: stripes and indexed column streams

ORC groups rows into stripes containing column streams, indexes, and a stripe footer. File-level metadata describes the file; row indexes and optional bloom filters help compatible readers skip smaller ranges. A bloom filter can rule out an absent value, but a positive result can be a false positive and still requires checking data. These features depend on writer settings and reader support. Hive’s transactional behavior requires its table and execution machinery; an ORC file alone is not a transaction system. ORC specification

Choose between ORC and Parquet using the engines that must consume the data, supported types, filter execution, and representative benchmarks. This lab writes both with Snappy, explicitly overriding PyArrow ORC’s uncompressed default. Matching the codec does not match all encodings, group sizes, or tuning. Exercise 6 verifies an ORC projection; it does not measure stripe pruning. ORC writer options

Avro: records and reader/writer schemas

Avro encodes a record’s fields in schema order. An object container file carries its writer schema and stores blocks of records with synchronization markers and an optional codec. Individual event payloads can also use Avro encoding, but their framing and access to the writer schema must be agreed with the consumer; a registry ID is an integration convention, not a mandatory prefix of every Avro datum.

During schema resolution the reader and writer schemas are matched. If the new reader expects a field absent from the writer schema, the reader’s default supplies its value; without that default, resolution fails. A nullable type alone does not supply a default. A reader default fills an absent field; it does not replace a null already written for that field. Reader-ignored fields, allowed type promotions, and aliases also have specified rules. Exercise 7 verifies a new nullable field with and without its default.

Record-oriented events suit a consumer that acts on each complete event. Avro container files can be split at block boundaries, but do not provide Parquet-style column chunks and min/max pruning. Selecting two fields still requires traversing records or encoded fields. Avro can be scanned analytically; repeated wide-table analytics are a reason to materialize a columnar copy. Avro specification

Arrow: arrays, IPC, and zero-copy conditions

Arrow arrays describe typed column buffers. Fixed-width values have a regular layout; strings use offsets and a data buffer; validity bitmaps represent nulls; nested arrays reference child arrays. Compatible libraries can share these buffers instead of constructing Python objects row by row. A Parquet read still decompresses and decodes storage encodings to produce Arrow arrays. That read is not automatically zero-copy. Buffer sharing depends on type, chunking, ownership, and the receiving API. Arrow columnar format

Arrow also defines IPC streams and seekable files. Feather V2 uses the IPC file format and supports optional LZ4 or Zstandard compression. An uncompressed memory-mapped file can expose compatible buffers without copying them into a second user-space allocation, but compression requires decompression and conversion can allocate new buffers. Flight transports Arrow batches between services; networking still has transport and serialization costs. Feather V2

The lab compares an uncompressed IPC file with Snappy Parquet and checks value equality. It does not establish that Arrow files are always larger or faster, nor that an archive must never use them. For a shared analytical lake, reader interoperability and filtering commonly favor Parquet or ORC. For a typed cache or array exchange, IPC can avoid repeated conversion work.

Turn the comparison into a file contract

For the subscription-company example, retain received payloads when replay requires exact originals; write typed analytical tables as Parquet where the consuming engines support it. Test Zstandard against Snappy on representative data. Record row-group rows, observed byte sizes, partition keys, sort columns, compaction thresholds, and supported readers. The lab’s small-file measurements motivate a test, not a universal 256 MB–1 GB rule.

Use Avro for event contracts when producers and consumers agree on schema resolution and framing. Use Arrow IPC for compatible caches or hand-offs. Preserve stable nested fields as structs and lists, and retain unknown payloads separately when needed. Before dropping a costly reference column, check debugging, joins, replay, and audit requirements. Projection already lets unrelated queries skip that column.

File formats do not define which files comprise an atomic table snapshot. Concurrent publication, schema governance, updates, and garbage collection still require a catalog, manifest, or table layer. A codec choice cannot repair inconsistent snapshots.

Lab

Run the setup once, then each exercise independently with that setup in scope. Tested with Python 3.12.14, PyArrow 25.0.1, and fastavro 1.12.2; install the latter two in a disposable environment. The million-row fixture uses integer cents in one assumed currency and naive timestamps interpreted as UTC for this exercise. It writes CSV, Snappy Parquet, and Snappy ORC; later exercises add IPC and a 1,000-record, two-field Avro sample. This is not an equal-size benchmark of all four formats. Allow several hundred MB of memory and temporary disk space. Sizes use decimal MB and reflect the recorded library/writer versions. Keep the TemporaryDirectory object alive through the exercises and call workspace.cleanup() after finishing. No cloud requests or timing claims are measured.

import os, random, tempfile
import pyarrow as pa
import pyarrow.compute as pc
import pyarrow.csv as pcsv
import pyarrow.orc as orc
import pyarrow.parquet as pq
from datetime import datetime, timedelta

rng = random.Random(0)
N = 1_000_000
start = datetime(2026, 1, 1)
rows = {
    "order_id": list(range(1, N + 1)),
    "customer_id": [rng.randint(1, 200_000) for _ in range(N)],
    "country": [rng.choice(["US", "US", "US", "KR", "DE", "JP", "BR"]) for _ in range(N)],
    "status": [rng.choice(["paid", "paid", "paid", "shipped", "refunded"]) for _ in range(N)],
    "amount_cents": [rng.randint(500, 20_000) for _ in range(N)],
    "order_ts": [start + timedelta(seconds=i * 7 + rng.randint(0, 6)) for i in range(N)],
    "note": [f"ref-{rng.randrange(10**9):09d}" for _ in range(N)],
}
table = pa.table(rows)
del rows
workspace = tempfile.TemporaryDirectory(prefix="de017-")
WORK = workspace.name

def path(name):
    return os.path.join(WORK, name)

def size_mb(name):
    return os.path.getsize(path(name)) / 1_000_000

pcsv.write_csv(table, path("orders.csv"))
pq.write_table(table, path("orders.parquet"), compression="snappy", row_group_size=100_000)
orc.write_table(table, path("orders.orc"), compression="snappy")
print(table.num_rows, "rows,", table.num_columns, "columns; schema:", ", ".join(f"{f.name}:{f.type}" for f in table.schema))
print(f"csv {size_mb('orders.csv'):.1f} MB, parquet {size_mb('orders.parquet'):.1f} MB, orc {size_mb('orders.orc'):.1f} MB")
# 1000000 rows, 7 columns; schema: order_id:int64, customer_id:int64, country:string, status:string, amount_cents:int64, order_ts:timestamp[us], note:string
# csv 75.2 MB, parquet 33.5 MB, orc 16.9 MB

1. Projection: sum compressed chunk sizes and verify selected values.

meta = pq.ParquetFile(path("orders.parquet")).metadata
per_column = {}
for g in range(meta.num_row_groups):
    for c in range(meta.num_columns):
        chunk = meta.row_group(g).column(c)
        per_column[chunk.path_in_schema] = per_column.get(chunk.path_in_schema, 0) + chunk.total_compressed_size
print("compressed bytes per column:", {k: f"{v / 1_000_000:.1f} MB" for k, v in per_column.items()})
wanted = ["order_id", "amount_cents"]
print(f"selected chunks {wanted}: {sum(per_column for c in wanted) / 1_000_000:.1f} MB of {sum(per_column.values()) / 1_000_000:.1f} MB")
projected = pq.read_table(path("orders.parquet"), columns=wanted)
print("projection equals input:", projected.equals(table.select(wanted)))
# compressed bytes per column: {'order_id': '6.0 MB', 'customer_id': '5.7 MB', 'country': '0.4 MB', 'status': '0.3 MB', 'amount_cents': '2.6 MB', 'order_ts': '8.2 MB', 'note': '10.2 MB'}
# selected chunks ['order_id', 'amount_cents']: 8.7 MB of 33.5 MB
# projection equals input: True

2. Filter candidates: compare sorted and shuffled files against an in-memory reference.

def candidate_groups(metadata, lo, hi):
    if lo >= hi:
        raise ValueError("expected lo < hi")
    index = metadata.schema.names.index("order_ts")
    candidates = []
    for g in range(metadata.num_row_groups):
        stats = metadata.row_group(g).column(index).statistics
        if stats is None or not stats.has_min_max or not (stats.max < lo or stats.min >= hi):
            candidates.append(g)
    return candidates

lo, hi = datetime(2026, 1, 20), datetime(2026, 1, 21)
filters = [("order_ts", ">=", lo), ("order_ts", "<", hi)]
cols = ["order_id", "amount_cents"]
expected = table.filter(pc.and_(pc.greater_equal(table["order_ts"], lo), pc.less(table["order_ts"], hi))).select(cols)
shuffled = table.take(pa.array(random.Random(1).sample(range(N), N)))
pq.write_table(shuffled, path("shuffled.parquet"), compression="snappy", row_group_size=100_000)
for name in ("orders.parquet", "shuffled.parquet"):
    meta = pq.ParquetFile(path(name)).metadata
    hits = pq.read_table(path(name), columns=cols, filters=filters)
    same = hits.sort_by("order_id").equals(expected.sort_by("order_id"))
    print(name, "candidate groups:", len(candidate_groups(meta, lo, hi)), "of", meta.num_row_groups,
          "matching rows:", hits.num_rows, "correct:", same)
# orders.parquet candidate groups: 1 of 10 matching rows: 12343 correct: True
# shuffled.parquet candidate groups: 10 of 10 matching rows: 12343 correct: True

3. Encoding and compression: compare codecs, then isolate the dictionary setting on one column.

for codec in ("none", "snappy", "zstd"):
    pq.write_table(table, path(f"orders_{codec}.parquet"), compression=codec, row_group_size=100_000)
    print(f"{codec:7} {size_mb(f'orders_{codec}.parquet'):5.1f} MB")
meta = pq.ParquetFile(path("orders_snappy.parquet")).metadata
for c in range(meta.num_columns):
    chunk = meta.row_group(0).column(c)
    print(f"  {chunk.path_in_schema:13} encodings {sorted(set(chunk.encodings))!s:40} {chunk.total_compressed_size / 1000:7.0f} KB per row group")
print("distinct country/note:", pc.count_distinct(table["country"]).as_py(), pc.count_distinct(table["note"]).as_py())
for use_dictionary in (False, True):
    name = f"country-{use_dictionary}.parquet"
    pq.write_table(table.select(["country"]), path(name), compression="snappy", row_group_size=100_000, use_dictionary=use_dictionary)
    print("country dictionary", use_dictionary, "bytes:", os.path.getsize(path(name)))
# none     50.7 MB
# snappy   33.5 MB
# zstd     24.1 MB
#   order_id      encodings ['PLAIN', 'RLE', 'RLE_DICTIONARY']           603 KB per row group
#   customer_id   encodings ['PLAIN', 'RLE', 'RLE_DICTIONARY']           574 KB per row group
#   country       encodings ['PLAIN', 'RLE', 'RLE_DICTIONARY']            38 KB per row group
#   status        encodings ['PLAIN', 'RLE', 'RLE_DICTIONARY']            26 KB per row group
#   amount_cents  encodings ['PLAIN', 'RLE', 'RLE_DICTIONARY']           264 KB per row group
#   order_ts      encodings ['PLAIN', 'RLE', 'RLE_DICTIONARY']           820 KB per row group
#   note          encodings ['PLAIN', 'RLE', 'RLE_DICTIONARY']          1024 KB per row group
# distinct country/note: 5 999488
# country dictionary False bytes: 1655960
# country dictionary True bytes: 381149

4. Schema evolution: add, rename, reject incompatible types, and allow deliberate widening.

import pyarrow.dataset as ds

v1 = table.slice(0, 1000)
v2 = table.slice(1000, 1000).append_column("channel", pa.array(["app"] * 1000))
files = [path("v1.parquet"), path("v2.parquet")]
for batch, name in zip((v1, v2), files):
    pq.write_table(batch, name)
unified = pa.unify_schemas([pq.read_schema(f) for f in files])
merged = ds.dataset(files, schema=unified, format="parquet").to_table()
print("unified:", merged.num_rows, "rows; channel nulls:", merged["channel"].null_count)
first_schema = ds.dataset(files, schema=v1.schema, format="parquet").to_table()
print("first schema:", first_schema.num_rows, "rows;", first_schema.num_columns, "columns; channel visible:", "channel" in first_schema.column_names)
v3 = v1.set_column(4, "amount_cents", pc.cast(v1["amount_cents"], pa.string()))
try:
    pa.unify_schemas([v1.schema, v3.schema])
except (pa.ArrowInvalid, pa.ArrowTypeError):
    print("int64/string conflict rejected")
small = pa.schema([("amount_cents", pa.int32())])
wide = pa.schema([("amount_cents", pa.int64())])
print("permissive integer promotion:", pa.unify_schemas([small, wide], promote_options="permissive").field("amount_cents").type)
renamed = v1.rename_columns(["nation" if n == "country" else n for n in v1.column_names])
pq.write_table(renamed, path("renamed.parquet"))
rename_schema = pa.unify_schemas([v1.schema, renamed.schema])
rename_result = ds.dataset([files[0], path("renamed.parquet")], schema=rename_schema, format="parquet").to_table()
print("name-based rename nulls:", rename_result["country"].null_count, rename_result["nation"].null_count)
# unified: 2000 rows; channel nulls: 1000
# first schema: 2000 rows; 7 columns; channel visible: False
# int64/string conflict rejected
# permissive integer promotion: int64
# name-based rename nulls: 1000 1000

5. File count: compare total and footer bytes for the same rows.

import pyarrow.dataset as ds

file_results = {}
for name, count in (("many", 1000), ("few", 10)):
    os.makedirs(path(name), exist_ok=True)
    per_file = N // count
    files = []
    for i in range(count):
        dest = path(f"{name}/part-{i:04d}.parquet")
        pq.write_table(table.slice(i * per_file, per_file), dest, compression="snappy", row_group_size=100_000)
        files.append(dest)
    total = sum(os.path.getsize(f) for f in files)
    footer = sum(pq.ParquetFile(f).metadata.serialized_size for f in files)
    result = ds.dataset(files, format="parquet").to_table(columns=["amount_cents"])
    correct = result.num_rows == N and pc.sum(result["amount_cents"]).as_py() == pc.sum(table["amount_cents"]).as_py()
    file_results[name] = (total, footer, correct)
    print(f"{count} files: {total / 1_000_000:.1f} MB; {footer / 1000:.0f} KB footers; aggregate correct: {correct}")
# 1000 files: 37.7 MB; 1586 KB footers; aggregate correct: True
# 10 files: 33.5 MB; 16 KB footers; aggregate correct: True

6. ORC and Arrow: verify projection, aggregation, and an IPC round trip.

import pyarrow.feather as feather

by_country = table.group_by("country").aggregate([("amount_cents", "sum"), ("order_id", "count")]).sort_by("country")
print(by_country.to_pydict())
orc_columns = orc.ORCFile(path("orders.orc")).read(columns=["order_id", "amount_cents"])
print("ORC projected values preserved:", orc_columns.equals(table.select(["order_id", "amount_cents"])))
feather.write_feather(table, path("orders.arrow"), compression="uncompressed", version=2)
arrow_copy = feather.read_table(path("orders.arrow"), memory_map=True)
print("Arrow round trip:", arrow_copy.equals(table))
print(f"uncompressed Arrow IPC {size_mb('orders.arrow'):.1f} MB; Snappy Parquet {size_mb('orders.parquet'):.1f} MB")
# {'country': ['BR', 'DE', 'JP', 'KR', 'US'], 'amount_cents_sum': [1455588946, 1469659158, 1457990804, 1468019477, 4398563982], 'order_id_count': [142189, 143317, 142558, 142590, 429346]}
# ORC projected values preserved: True
# Arrow round trip: True
# uncompressed Arrow IPC 64.4 MB; Snappy Parquet 33.5 MB

7. Avro reader defaults: read old records with a new schema.

from fastavro import writer, reader
from copy import deepcopy

writer_schema = {"type": "record", "name": "Order", "fields": [
    {"name": "order_id", "type": "long"},
    {"name": "amount_cents", "type": "long"}]}
reader_schema = deepcopy(writer_schema)
reader_schema["fields"].append({"name": "channel", "type": ["null", "string"], "default": None})
records = table.select(["order_id", "amount_cents"]).slice(0, 1000).to_pylist()
with open(path("orders.avro"), "wb") as stream:
    writer(stream, writer_schema, records, codec="null")
with open(path("orders.avro"), "rb") as stream:
    decoded = list(reader(stream, reader_schema=reader_schema))
print("Avro rows:", len(decoded), "default nulls:", sum(r["channel"] is None for r in decoded))
print("original fields preserved:", [{k: r[k] for k in ("order_id", "amount_cents")} for r in decoded] == records)
bad_reader = deepcopy(reader_schema)
del bad_reader["fields"][-1]["default"]
from fastavro._read_common import SchemaResolutionError
try:
    with open(path("orders.avro"), "rb") as stream:
        list(reader(stream, reader_schema=bad_reader))
except SchemaResolutionError:
    print("missing reader default rejected")
# Avro rows: 1000 default nulls: 1000
# original fields preserved: True
# missing reader default rejected

8. Nested values: preserve null, empty, and repeated structures.

item_type = pa.struct([("sku", pa.string()), ("qty", pa.int64())])
nested_schema = pa.schema([("order_id", pa.int64()), ("items", pa.list_(item_type))])
nested = pa.Table.from_pylist([
    {"order_id": 1, "items": None},
    {"order_id": 2, "items": []},
    {"order_id": 3, "items": [None]},
    {"order_id": 4, "items": [{"sku": "A", "qty": None}, {"sku": "B", "qty": 2}]}
], schema=nested_schema)
pq.write_table(nested, path("nested.parquet"))
restored = pq.read_table(path("nested.parquet"))
print("nested round trip:", restored.equals(nested))
print(restored["items"].to_pylist())
# nested round trip: True
# [None, [], [None], [{'sku': 'A', 'qty': None}, {'sku': 'B', 'qty': 2}]]

9. Write and challenge a file standard.