10 - Serialization Formats - Python

28 min read

Serialization Formats - Python

Quote

“Write programs to handle text streams, because that is a universal interface.”

— Doug McIlroy, Bell System Technical Journal (1978)

import os
import csv
import json
import struct
import tempfile
import time
import shutil
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import pyarrow.compute as pc
from pathlib import Path
from io import BytesIO
from google.protobuf import descriptor_pb2, descriptor_pool
from google.protobuf import message as _message
from google.protobuf import reflection as _reflection
 
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from IPython.display import display
import fastavro
import random
html_formatter = get_ipython().display_formatter.formatters['text/html'] # type: ignore
html_formatter.for_type(pd.DataFrame, lambda df: df.to_html())
html_formatter.for_type(pd.Series, lambda s: s.to_frame().to_html())

Parquet Files

pyarrow | write and read Parquet

Parquet overview — columnar format for analytics

Parquet stores data column-by-column with per-column compression (snappy, gzip, zstd). Schema is embedded in the file footer — self-describing, no separate schema file needed. Supports column pruning (projection pushdown), predicate pushdown, and partitioning. 5-10x smaller than CSV. Standard in data lakes (GCS, S3, ADLS), BigQuery, Spark, DuckDB, Athena. Use pyarrow for parquet I/O in Python.

When NOT to use Parquet

  • Small files (<1MB) — Parquet overhead exceeds benefit
  • Frequent appends — Parquet is immutable; use Avro/JSONL for streaming
  • Simple data exchange — CSV is more universal

Use Parquet for analytics; JSONL or Avro for streaming appends

For batch analytics (BigQuery, Spark, Athena, DuckDB) with files >1 MB: Parquet with snappy compression. For event streams where records arrive continuously: JSONL (one object per line) or Avro with a schema registry. For simple hand-off to non-engineering consumers: CSV.

tmp_dir = Path(tempfile.mkdtemp(prefix="parquet_"))

pa.schema — define typed Arrow schema

pa.schema() declares column names and types upfront. Arrow enforces these types at table creation — no silent coercion. Use pa.string(), pa.int64(), pa.float64(), pa.bool_() for the most common column types.

schema = pa.schema([
    ("event_id", pa.string()),
    ("event_type", pa.string()),
    ("user_id", pa.int64()),
    ("revenue", pa.float64()),
    ("is_mobile", pa.bool_()),
])

pa.table — create Arrow table from column arrays

Each column is a pa.array — typed, nullable, and memory-efficient. Pass a dict of column names to Python lists along with the schema to enforce types at construction time.

table = pa.table({
    "event_id":   ["evt_001", "evt_002", "evt_003", "evt_004", "evt_005"],
    "event_type": ["page_view", "purchase", "page_view", "signup", "purchase"],
    "user_id":    [1001, 1002, 1001, 1003, 1002],
    "revenue":    [0.0, 49.99, 0.0, 0.0, 129.99],
    "is_mobile":  [True, False, True, True, False],
}, schema=schema)

pq.write_table — write Parquet from Arrow table

pq.write_table() serializes an Arrow table to a Parquet file on disk. The compression parameter controls the codec: 'snappy' (default, fast), 'gzip' (smaller), 'zstd' (best ratio), 'none'.

parquet_file = tmp_dir / "events.parquet"
pq.write_table(table, parquet_file, compression="snappy")
 
print(parquet_file.name)
print(f"{parquet_file.stat().st_size} bytes (compressed)")
print(f"Rows: {table.num_rows}, Columns: {table.num_columns}")
events.parquet
1594 bytes (compressed)
Rows: 5, Columns: 5

pyarrow pq.read_table — read entire Parquet file

pq.read_table() reads the full Parquet file into an Arrow table. The schema is discovered from the file footer. Call .to_pandas() to convert to a pandas DataFrame for display or further processing.

table_read = pq.read_table(parquet_file)
print(f"Schema:\n{table_read.schema}")
print(f"Data:\n{table_read.to_pandas()}")
Schema:
  event_id: string
  event_type: string
  user_id: int64
  revenue: double
  is_mobile: bool
 
Data:
  event_id event_type  user_id  revenue  is_mobile
  evt_001  page_view     1001     0.00       True
  evt_002   purchase     1002    49.99      False
  evt_003  page_view     1001     0.00       True
  evt_004     signup     1003     0.00       True
  evt_005   purchase     1002   129.99      False

Column pruning, pushdown, and partitioning

pyarrow pq.read_table columns= — column pruning

Pass columns=["col1", "col2"] to read only the columns you need. On a 100-column table, this can be 50x faster than CSV because Parquet stores columns independently — unneeded columns are never read from disk.

partial = pq.read_table(parquet_file, columns=["event_id", "revenue"])
print(partial.column_names)
print(partial.column('revenue').to_pylist())
print(f"{sum(partial.column('revenue').to_pylist()):.2f}")
['event_id', 'revenue']
[0.0, 49.99, 0.0, 0.0, 129.99]
179.98

pyarrow pq.read_table filters= — predicate pushdown

Pass filters=[("column", "op", "value")] to skip entire row groups whose statistics prove no matching rows exist. This avoids reading data that would be filtered out anyway — significant for partitioned datasets with millions of rows.

filtered = pq.read_table(
    parquet_file,
    filters=[("event_type", "==", "purchase")]
)
print(f"Purchases only ({filtered.num_rows} rows):")
print(f"{filtered.to_pandas()}")
 
meta = pq.read_metadata(parquet_file)
print(f"Rows: {meta.num_rows}, Columns: {meta.num_columns}, Row groups: {meta.num_row_groups}")
 
schema_read = pq.read_schema(parquet_file)
for i, field in enumerate(schema_read):
    print(f"    [{i}] {field.name}: {field.type}")

Metadata reads the schema and row count from the file footer without loading any data — instant even for huge files.

Purchases only (2 rows):
  event_id event_type  user_id  revenue  is_mobile
  evt_002   purchase     1002    49.99      False
  evt_005   purchase     1002   129.99      False
Rows: 5, Columns: 5, Row groups: 1
Schema:
  [0] event_id: string
  [1] event_type: string
  [2] user_id: int64
  [3] revenue: double
  [4] is_mobile: bool

pyarrow pq.write_to_dataset — Hive-style partitioning

pq.write_to_dataset() splits a table into subdirectories by partition column values (e.g., event_type=purchase/). Readers discover partitions automatically. This is the standard storage pattern for data lakes — BigQuery, Athena, and Spark all understand Hive-style layout.

partitioned_dir = tmp_dir / "events_partitioned"
pq.write_to_dataset(
    table,
    root_path=str(partitioned_dir),
    partition_cols=["event_type"],
)
 
for f in sorted(partitioned_dir.rglob("*.parquet")):
    rel = f.relative_to(partitioned_dir)
    print(f"    {rel} ({f.stat().st_size} bytes)")
 
dataset = pq.read_table(str(partitioned_dir))
print(f"{dataset.num_rows} rows, columns: {dataset.column_names}")

pyarrow discovers partitions automatically when reading back.

event_type=page_view/...-0.parquet (1266 bytes)
event_type=purchase/...-0.parquet (1274 bytes)
event_type=signup/...-0.parquet (1255 bytes)
5 rows, columns: ['event_id', 'user_id', 'revenue', 'is_mobile', 'event_type']

In-memory and comparison

Parquet in memory — BytesIO | cloud upload without temp files

Serialize Parquet to a BytesIO buffer for direct cloud upload (GCS, S3) without writing to disk. Rewind with seek(0) before reading or uploading.

buffer = BytesIO()
pq.write_table(table, buffer)
print(f"{buffer.tell()} bytes")
 
buffer.seek(0)
table_from_mem = pq.read_table(buffer)
print(f"{table_from_mem.num_rows} rows")
1594 bytes
5 rows

CSV vs Parquet comparison | size, features, and use cases

csv_file = tmp_dir / "events.csv"
table.to_pandas().to_csv(csv_file, index=False)
 
csv_size = csv_file.stat().st_size
parquet_size = parquet_file.stat().st_size
print(f"{csv_size} bytes")
print(f"{parquet_size} bytes")
print(f"Ratio:        {csv_size / parquet_size:.1f}x smaller with parquet")
214 bytes
1594 bytes
0.1x smaller with parquet
FeatureCSVParquet
FormatText (row-based)Binary (columnar)
SchemaNo (header row only)Embedded (typed, nullable)
CompressionNone (manual gzip)Built-in (snappy/gzip/zstd)
Column pruningNo (read all columns)Yes (read only what you need)
PartitioningManual (directory naming)Native (Hive-style)
Use caseSimple exchange, legacyData lakes, analytics, BigQuery

Enterprise Message Serialization

Binary formats provide schema enforcement, cross-language support, and compact serialization (3–10x smaller than JSON).

Apache Avro — binary format with embedded schema (self-describing). Schema evolution lets you add/remove fields without breaking consumers. Standard for Kafka messages in data engineering. Python library: fastavro. ~50–70% smaller than JSON.

Protocol Buffers (Protobuf) — binary format with separate .proto schema files. protoc compiles schemas into Python/Java/Go/C# classes. Standard for gRPC microservices. ~60–80% smaller than JSON.

Why not JSON or pickle

Why not JSON or pickle at scale?

  • JSON: text-based, no schema enforcement, slow to parse at scale
  • pickle: Python-only, insecure (arbitrary code execution), no schema — never use in production

Use Avro for Kafka, Protobuf for gRPC — both enforce schema

Avro stores the schema in the file/message header — consumers always know the shape of the data. Protobuf uses compiled .proto files — ideal for gRPC where both sides share the generated code. Both are cross-language (Python, Java, Go, C#) and 50-80% smaller than equivalent JSON.

FormatSizeSpeedSchemaCross-langUse case
JSONLargeSlowNoYesAPIs, config
AvroSmallFastYesYesKafka, data lakes
ProtobufSmallFastestYesYesgRPC, mobile
pickleMediumFastNoNoNever in prod
structTinyFastestManualManualIoT, binary protocols

fastavro | Apache Avro

Avro serialization with fastavro | define schema and parse

Avro schemas are defined as JSON dicts with type, name, fields. fastavro.parse_schema() validates the schema. Records are plain Python dicts — fastavro validates them against the schema on write.

avro_schema = {
    "type": "record",
    "name": "StockQuote",
    "namespace": "stoxx",
    "fields": [
        {"name": "symbol",   "type": "string"},
        {"name": "price",    "type": "double"},
        {"name": "volume",   "type": "long"},
        {"name": "exchange", "type": ["null", "string"], "default": None},
    ],
}
 
parsed_schema = fastavro.parse_schema(avro_schema)
print(f"{avro_schema['name']} ({len(avro_schema['fields'])} fields)")
for f in avro_schema["fields"]:
    print(f"    {f['name']}: {f['type']}")
StockQuote (4 fields)
  symbol: string
  price: double
  volume: long
  exchange: ['null', 'string']

Write and read Avro file | fastavro.writer with embedded schema

fastavro.writer embeds the schema in the file header and writes records as compact binary. fastavro.reader discovers the schema from the file header automatically — no external schema file needed.

avro_dir = tempfile.mkdtemp(prefix="avro_py_")
avro_file = os.path.join(avro_dir, "quotes.avro")
 
records = [
    {"symbol": "SAP.DE",  "price": 166.52, "volume": 82621,  "exchange": "XETR"},
    {"symbol": "ASML.AS", "price": 685.40, "volume": 45000,  "exchange": "XAMS"},
    {"symbol": "TTE.PA",  "price": 58.20,  "volume": 120000, "exchange": "XPAR"},
    {"symbol": "BAS.DE",  "price": 44.85,  "volume": 95000,  "exchange": None},
]
 
with open(avro_file, "wb") as f:
    fastavro.writer(f, parsed_schema, records)
 
print(os.path.basename(avro_file))
print(f"Records: {len(records)}, Size: {os.path.getsize(avro_file)} bytes")
 
with open(avro_file, "rb") as f:
    reader = fastavro.reader(f)
    print(f"  Schema from file: {reader.writer_schema['name']}")  # type: ignore[index]
    for record in reader:
        exch = record["exchange"] or "N/A"  # type: ignore[index]
        print(f"    {record["symbol"]:10} €{record["price"]:>8.2f}  vol={record["volume"]:>8}  exch={exch}")  # type: ignore[index]
 
shutil.rmtree(avro_dir)

Records are plain Python dicts — fastavro validates them against the schema on write. The schema is read from the file header automatically on read.

quotes.avro
Records: 4, Size: 399 bytes
Schema from file: stoxx.StockQuote
  SAP.DE     €  166.52  vol=   82621  exch=XETR
  ASML.AS    €  685.40  vol=   45000  exch=XAMS
  TTE.PA     €   58.20  vol=  120000  exch=XPAR
  BAS.DE     €   44.85  vol=   95000  exch=N/A

Avro in memory and schema evolution | BytesIO for Kafka payloads

Serialize Avro records to BytesIO for Kafka producer payloads or API responses without disk I/O. Schema evolution rules: adding a field with a default is safe (backward compatible); changing a field type breaks consumers.

avro_buffer = BytesIO()
fastavro.writer(avro_buffer, parsed_schema, records)
avro_bytes = avro_buffer.getvalue()
print(f"{len(avro_bytes)} bytes ({len(records)} records)")
 
avro_buffer.seek(0)
mem_records = list(fastavro.reader(avro_buffer))
print(f"{len(mem_records)} records")
 
json_size = len(json.dumps(records).encode())
print(f"{json_size} bytes")
print(f"{len(avro_bytes)} bytes")
print(f"Savings:    {(1 - len(avro_bytes)/json_size)*100:.0f}%")
 
399 bytes (4 records)
4 records
JSON: 300 bytes
Avro: 399 bytes
Savings: -33%

Schema evolution rules

  • Safe: add field with default, remove field with default, add aliases
  • Unsafe: change field type, remove field WITHOUT default

protobuf | Protocol Buffers

Protobuf in Python — dynamic message building

In notebooks or dynamic scenarios without protoc-generated classes, use descriptor_pb2 to define the schema at runtime and _reflection.GeneratedProtocolMessageType to create a message class. SerializeToString() produces the same binary format as protoc-generated code.

DESCRIPTOR = descriptor_pb2.FileDescriptorProto(
    name="stock_quote.proto",
    package="stoxx",
    message_type=[
        descriptor_pb2.DescriptorProto(
            name="StockQuote",
            field=[
                descriptor_pb2.FieldDescriptorProto(
                    name="symbol", number=1,
                    type=descriptor_pb2.FieldDescriptorProto.TYPE_STRING,
                    label=descriptor_pb2.FieldDescriptorProto.LABEL_OPTIONAL,
                ),
                descriptor_pb2.FieldDescriptorProto(
                    name="price", number=2,
                    type=descriptor_pb2.FieldDescriptorProto.TYPE_DOUBLE,
                    label=descriptor_pb2.FieldDescriptorProto.LABEL_OPTIONAL,
                ),
                descriptor_pb2.FieldDescriptorProto(
                    name="volume", number=3,
                    type=descriptor_pb2.FieldDescriptorProto.TYPE_INT64,
                    label=descriptor_pb2.FieldDescriptorProto.LABEL_OPTIONAL,
                ),
            ],
        ),
    ],
)
 
pool = descriptor_pool.DescriptorPool()
file_desc = pool.Add(DESCRIPTOR)
msg_desc = pool.FindMessageTypeByName("stoxx.StockQuote")
 
factory = _reflection.GeneratedProtocolMessageType(
    "StockQuote",
    (_message.Message,),
    {"DESCRIPTOR": msg_desc, "__module__": "__main__"},
)
 
quote = factory(symbol="SAP.DE", price=166.52, volume=82621)  # type: ignore[call-arg]
binary = quote.SerializeToString()  # type: ignore[attr-defined]
print(f"Message:    symbol={quote.symbol}, price={quote.price}, volume={quote.volume}")  # type: ignore[attr-defined]
print(f"{len(binary)} bytes ({binary.hex()[:40]}...)")
 
parsed = factory.FromString(binary)  # type: ignore[attr-defined]
print(f"Parsed:     symbol={parsed.symbol}, price={parsed.price}, volume={parsed.volume}")  # type: ignore[attr-defined]
 
json_size = len(json.dumps({"symbol": "SAP.DE", "price": 166.52, "volume": 82621}).encode())
print(f"{json_size} bytes")
print(f"{len(binary)} bytes")
print(f"Savings:       {(1 - len(binary)/json_size)*100:.0f}%")
Message: symbol=SAP.DE, price=166.52, volume=82621
21 bytes (0a065341502e444511713d0ad7a3d0644018bd85...)
Parsed:  symbol=SAP.DE, price=166.52, volume=82621
JSON:     54 bytes
Protobuf: 21 bytes
Savings:  61%

Protobuf with protoc (production pattern) | protoc-generated code workflow

In production, define schemas in .proto files, compile with protoc --python_out=. to generate _pb2.py modules, then serialize/deserialize with SerializeToString() and FromString(). Schema evolution: old code ignores new fields; new code uses defaults for missing fields.

Step 1 — Define schema (stock_quote.proto). Field numbers are wire identifiers, not default values. Adding new fields (like exchange = 4) is backward-compatible — old consumers ignore unknown fields.

syntax = "proto3";
package stoxx;
 
message StockQuote {
  string symbol = 1;
  double price = 2;
  int64  volume = 3;
  string exchange = 4;
}

Step 2 — Compile with protoc to generate a Python module. The generated _pb2.py file contains typed classes with SerializeToString() and FromString() methods.

protoc --python_out=. stock_quote.proto

This generates stock_quote_pb2.py with typed classes.

Step 3 — Use in Python. Serialize with SerializeToString() (returns bytes for Kafka, gRPC, or file storage) and deserialize with FromString(). Schema evolution is automatic — old code ignores field 4 (exchange), new code uses the default ("") if absent.

from stock_quote_pb2 import StockQuote
 
quote = StockQuote(symbol="SAP.DE", price=166.52, volume=82621)
data = quote.SerializeToString()
 
parsed = StockQuote.FromString(data)
print(parsed.symbol, parsed.price)

Format Performance Benchmark

For the architecture-level decision guide on when to use each format across the full pipeline (ingestion, storage, interchange), see serialization-formats. The benchmarks below focus on Python-specific library performance, while data-loading-and-export covers how format choice affects BigQuery load throughput.

Test data and benchmark setup

Generate synthetic OHLCV test data — three sizes for benchmarks

Generates synthetic OHLCV (open/high/low/close/volume) data matching the stoxx database schema. Three sizes — 100 (small), 10K (medium), 100K (large) — to show how format overhead scales.

random.seed(42)
 
symbols = ["SAP.DE","ASML.AS","TTE.PA","BAS.DE","BAYN.DE","BMW.DE","SIE.DE","ALV.DE",
           "ADS.DE","DTE.DE","ENEL.MI","ENI.MI","BNP.PA","MC.PA","OR.PA","AIR.PA",
           "SAN.PA","CS.PA","DG.PA","SU.PA","INGA.AS","AD.AS","PRX.AS","WKL.AS",
           "UCG.MI","ISP.MI","RACE.MI","ABI.BR","ARGX.BR","ITX.MC"]
 
def generate_data(count: int) -> list[dict]:
    data = []
    for i in range(count):
        close = 50 + random.random() * 200
        opn = close * (0.98 + random.random() * 0.04)
        high = max(opn, close) * (1 + random.random() * 0.02)
        low = min(opn, close) * (1 - random.random() * 0.02)
        data.append({
            "symbol": symbols[i % len(symbols)],
            "date": f"2020-{(i // 30 % 12) + 1:02d}-{(i % 28) + 1:02d}",
            "open": round(opn, 2), "high": round(high, 2),
            "low": round(low, 2), "close": round(close, 2),
            "volume": int(random.random() * 500_000),
        })
    return data
 
small  = generate_data(100)
medium = generate_data(10_000)
large  = generate_data(100_000)
print(f"Small: {len(small):,}, Medium: {len(medium):,}, Large: {len(large):,}")
Small: 100, Medium: 10,000, Large: 100,000

Benchmark execution

Format benchmarks — write/read speed across CSV, Parquet, Avro, Protobuf

Benchmarks each format at all three sizes, measuring write time (ms), read time (ms), and file size (bytes). Each format uses its standard Python library: csv module, json module, pandas/pyarrow for Parquet, fastavro for Avro, and raw wire-format encoding for Protobuf.

 
bench_dir = tempfile.mkdtemp(prefix="bench_py_")
results = []
 
def bench(fmt, label, data, write_fn, read_fn):
    path = os.path.join(bench_dir, f"{fmt}_{label}")
    start = time.perf_counter()
    write_fn(path, data)
    write_ms = (time.perf_counter() - start) * 1000
    file_bytes = os.path.getsize(path)
    start = time.perf_counter()
    count = read_fn(path)
    read_ms = (time.perf_counter() - start) * 1000
    results.append((fmt, label, len(data), write_ms, read_ms, file_bytes))
 
def write_csv(path, data):
    with open(path, "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=data[0].keys())
        w.writeheader(); w.writerows(data)
def read_csv(path):
    with open(path) as f: return sum(1 for _ in csv.DictReader(f))
 
def write_json(path, data):
    with open(path, "w") as f: json.dump(data, f)
def read_json(path):
    with open(path) as f: return len(json.load(f))
 
def write_parquet(path, data):
    pd.DataFrame(data).to_parquet(path, index=False)
def read_parquet(path):
    return len(pq.read_table(path))
 
bench_avro_schema = fastavro.parse_schema({
    "type": "record", "name": "Ohlcv", "fields": [
        {"name": "symbol", "type": "string"}, {"name": "date", "type": "string"},
        {"name": "open", "type": "double"}, {"name": "high", "type": "double"},
        {"name": "low", "type": "double"}, {"name": "close", "type": "double"},
        {"name": "volume", "type": "long"},
    ]
})
def write_avro(path, data):
    with open(path, "wb") as f: fastavro.writer(f, bench_avro_schema, data)
def read_avro(path):
    with open(path, "rb") as f: return sum(1 for _ in fastavro.reader(f))
 
def _proto_tag(field_num, wire_type):
    return bytes([(field_num << 3) | wire_type])
 
def _proto_varint(value):
    out = b''
    while value > 0x7f:
        out += bytes([value & 0x7f | 0x80])
        value >>= 7
    out += bytes([value & 0x7f])
    return out
 
def write_proto(path, data):
    with open(path, 'wb') as f:
        for r in data:
            msg = b''
            sym = r['symbol'].encode('utf-8')
            msg += _proto_tag(1, 2) + bytes([len(sym)]) + sym
            dt = r['date'].encode('utf-8')
            msg += _proto_tag(2, 2) + bytes([len(dt)]) + dt
            for fnum, key in [(3,'open'),(4,'high'),(5,'low'),(6,'close')]:
                msg += _proto_tag(fnum, 1) + struct.pack('<d', r[key])
            msg += _proto_tag(7, 0) + _proto_varint(r['volume'])
            f.write(len(msg).to_bytes(2, 'big') + msg)
 
def read_proto(path):
    count = 0
    with open(path, 'rb') as f:
        while True:
            lb = f.read(2)
            if len(lb) < 2: break
            f.read(int.from_bytes(lb, 'big'))
            count += 1
    return count
 
for label, data in [("small", small), ("medium", medium), ("large", large)]:
    for fmt, wfn, rfn, ext in [
        ("CSV", write_csv, read_csv, ".csv"),
        ("JSON", write_json, read_json, ".json"),
        ("Parquet", write_parquet, read_parquet, ".parquet"),
        ("Avro", write_avro, read_avro, ".avro"),
        ("Protobuf", write_proto, read_proto, ".bin"),
    ]:
        bench(fmt, label + ext, data, wfn, rfn)

Results and analysis

Benchmark results — write/read performance matrix

 
df_bench = pd.DataFrame(results, columns=["Format", "File", "Records", "Write_ms", "Read_ms", "Bytes"])
df_bench["Bucket"] = df_bench["File"].str.extract(r"(small|medium|large)")
df_bench["File_Size"] = df_bench["Bytes"].apply(
    lambda b: f"{b} B" if b < 1024 else f"{b/1024:.1f} KB" if b < 1024**2 else f"{b/1024**2:.1f} MB")
df_bench["Bytes/Rec"] = df_bench["Bytes"] // df_bench["Records"]
df_bench["Write_ms"] = df_bench["Write_ms"].round(0).astype(int)
df_bench["Read_ms"] = df_bench["Read_ms"].round(0).astype(int)
 
bucket_order = {"large": 0, "medium": 1, "small": 2}
df_out = (df_bench[["Format", "Records", "File_Size", "Bytes/Rec", "Write_ms", "Read_ms", "Bucket"]]
    .assign(_sort=df_bench["Bucket"].map(bucket_order))
    .sort_values(["_sort", "Bytes/Rec"]).drop(columns="_sort").reset_index(drop=True))
 
rows = []
prev_bucket = None
for _, row in df_out.iterrows():
    if row["Bucket"] != prev_bucket:
        sep = pd.Series({c: "" for c in df_out.columns})
        sep["Format"] = f"\u2501\u2501 {row['Records']:,} records \u2501\u2501"
        rows.append(sep)
        prev_bucket = row["Bucket"]
    rows.append(row)
 
df_display = pd.DataFrame(rows).reset_index(drop=True)
 
def highlight(df):
    styles = pd.DataFrame("", index=df.index, columns=df.columns)
    # Find separator row indices to define groups
    sep_idxs = df.index[df["Records"] == ""].tolist() + [len(df)]
    for g in range(len(sep_idxs) - 1):
        start = sep_idxs[g] + 1
        end = sep_idxs[g + 1]
        group = df.iloc[start:end]
        for col in ["Bytes/Rec", "Write_ms", "Read_ms"]:
            vals = pd.to_numeric(group[col], errors="coerce")
            positive = vals[vals > 0]
            if len(positive) > 0:
                styles.iloc[positive.idxmin(), styles.columns.get_loc(col)] = "background-color:#2e7d32;color:#fff"
                styles.iloc[vals.idxmax(), styles.columns.get_loc(col)] = "background-color:#c62828;color:#fff"
    for idx in sep_idxs[:-1]:
        for col in styles.columns:
            styles.iloc[idx, styles.columns.get_loc(col)] = "font-weight:bold;border-top:2px solid #888"
    return styles
 
display(
    df_display.drop(columns="Bucket")
    .style.apply(highlight, axis=None)
    .set_caption("Format Performance Benchmark — green = best, red = worst per bucket")
    .hide(axis="index")
)
Format Performance Benchmark — green = best, red = worst per bucket
FormatRecordsFile_SizeBytes/RecWrite_msRead_ms
━━ 100,000 records ━━
Parquet1000001.7 MB17708
CSV1000005.0 MB52218115
Avro1000005.1 MB53123102
Protobuf1000005.9 MB6118421
JSON10000011.9 MB124716102
━━ 10,000 records ━━
Parquet10000280.9 KB28136
CSV10000511.7 KB522522
Avro10000518.5 KB531218
Protobuf10000605.1 KB61235
JSON100001.2 MB1246713
━━ 100 records ━━
CSV1005.1 KB5211
Avro1005.5 KB5605
Protobuf1006.1 KB6101
Parquet1008.3 KB8435
JSON10012.1 KB12416

Write vs Read speed chart — Plotly grouped bar

 
colors = {"CSV": "#4285F4", "JSON": "#FBBC05", "Parquet": "#34A853", "Avro": "#EA4335", "Protobuf": "#9C27B0"}
large = df_bench[df_bench["Bucket"] == "large"].sort_values("Bytes")
 
fig = make_subplots(rows=1, cols=2, subplot_titles=("Write Time (ms)", "Read Time (ms)"))
for _, row in large.iterrows():
    fig.add_trace(go.Bar(name=row["Format"], x=[row["Format"]], y=[row["Write_ms"]],
                         marker_color=colors.get(row["Format"], "#999"), showlegend=False), row=1, col=1)
    fig.add_trace(go.Bar(name=row["Format"], x=[row["Format"]], y=[row["Read_ms"]],
                         marker_color=colors.get(row["Format"], "#999"), showlegend=False), row=1, col=2)
 
fig.update_layout(title_text="Read/Write Performance — 100K Records",
                  height=400, template="plotly_dark", bargap=0.3)
fig.show()

Compression comparison — file size per format for large tier

 
csv_bytes = df_bench[(df_bench["Format"] == "CSV") & (df_bench["Bucket"] == "large")]["Bytes"].iloc[0]
comp = df_bench[df_bench["Bucket"] == "large"][["Format", "Bytes"]].sort_values("Bytes").copy()
comp["vs_CSV"] = ((1 - comp["Bytes"] / csv_bytes) * 100).round(1).apply(lambda x: f"{x:+.1f}%")
comp["File_Size"] = comp["Bytes"].apply(
    lambda b: f"{b/1024**2:.1f} MB" if b >= 1024**2 else f"{b/1024:.1f} KB")
comp["Size_MB"] = (comp["Bytes"] / 1024**2).round(2)
 
display(
    comp[["Format", "File_Size", "vs_CSV"]]
    .style
    .set_caption("Compression vs CSV — 100K records (large bucket)")
    .hide(axis="index")
)
Compression vs CSV — 100K records (large bucket)
FormatFile_Sizevs_CSV
Parquet1.7 MB+65.9%
CSV5.0 MB+0.0%
Avro5.1 MB-1.3%
Protobuf5.9 MB-18.3%
JSON11.9 MB-137.5%

File size bar chart — Plotly horizontal bars

 
colors_list = [colors.get(f, "#999") for f in comp["Format"]]
fig2 = go.Figure(go.Bar(
    x=comp["Size_MB"].values,
    y=comp["Format"].values,
    orientation="h",
    marker_color=colors_list,
    text=comp.apply(lambda r: f"{r['File_Size']} ({r['vs_CSV']})", axis=1),
    textposition="outside",
))
fig2.update_layout(
    title="File Size Comparison — 100K Records",
    xaxis_title="Size (MB)",
    yaxis=dict(autorange="reversed"),
    height=350,
    template="plotly_dark",
    margin=dict(r=120),
)
fig2.show()

Recommendation matrix | best format for each scenario

ScenarioBest FormatWhy
Data lake / analytics queriesParquetColumnar: read 2 of 50 columns = skip 96% of data
Kafka event streamingAvroSchema embedded, schema registry, compact
gRPC microservicesProtobufFastest parse, smallest size, code-generated types
REST API responsesJSONHuman-readable, universal, self-describing
Config filesJSON / YAMLHuman-editable, comments (YAML), versioned in git
Legacy data warehouse exportCSVUniversal, every tool reads it
Debug / loggingJSONHuman-readable, structured, grep-friendly
High-freq trading feedProtobufLowest latency, smallest payload
ML feature storeParquetColumn pruning, predicate pushdown, partitioned
Cross-language IPCProtobuf/AvroProtobuf for speed, Avro for schema
Batch ETL intermediateParquetCompressed, typed, Spark/BigQuery/Polars
Small config payloads (<1KB)JSONBinary format overhead not worth it
Browser / mobile APIJSON + gzipUniversal client support

Rankings

Size (smallest → largest): Protobuf < Parquet < Avro < CSV < JSON Speed (fastest → slowest): Protobuf > CSV > Parquet > Avro > JSON

shutil.rmtree(bench_dir)

Python Serialization Formats Warnings

Parquet is not streamable

Parquet requires reading the footer (at the end of the file) before accessing any data. You cannot write partial Parquet files or stream rows one at a time.

Correct pattern

Use Avro for streaming/append workloads. Use Parquet for batch analytics where the entire file is written and read as a unit.

Protobuf schema not embedded in data

Unlike Avro and Parquet, Protobuf messages contain no schema — the receiver must have the .proto file to decode. Schema mismatch produces silent data corruption.

Correct pattern

Use a schema registry (Confluent, Buf) for versioned schemas. Always test backward/forward compatibility before deploying schema changes.

Avro schema changes can break consumers

Removing a field without a default value, or changing a field’s type, breaks existing readers.

Correct pattern

Always add new fields with default values. Mark removed fields as deprecated rather than deleting. Use Avro’s schema resolution rules for forward/backward compatibility.

Python Serialization Formats Recommendations

  • Use Parquet for data lakes and analytical storage — columnar layout, compression, predicate pushdown, schema embedded.
  • Use Avro for Kafka and streaming — schema embedded, schema evolution, compact, streaming-friendly.
  • Use Protobuf for gRPC and low-latency IPC — smallest payload, fastest parse, strongly typed.
  • Use JSON for APIs and human-readable interchange — universal support, self-describing, debuggable.
  • Use CSV only for legacy interchange — no types, no schema, no nesting, no compression. Prefer Parquet for anything analytical.
  • Choose compression by workload — Snappy for speed (real-time), Zstd for balance (ETL), Gzip for maximum compression (archival).
  • Test schema evolution before deploying — add fields with defaults, never rename, use compatibility checks.

Python Serialization Formats Troubleshooting

ProblemCauseFix
Parquet read only returns some columnsColumn pruning is working as intendedSpecify all needed columns in columns= parameter
Avro deserialization fails after schema changeIncompatible schema evolution (removed required field)Add default values to new fields; don’t remove required fields
Protobuf message is empty/zero-lengthSerialized with wrong schema or all fields are default valuesCheck .proto field numbers match; Protobuf omits default values by design
Parquet file much larger than expectedWrong compression codec or no compressionSet compression='zstd' or compression='snappy'
fastavro raises SchemaParseExceptionAvro schema JSON is malformedValidate schema against Avro spec; check field types and names

Graph View

Backlinks