pybufarrow

Turn raw protobuf bytes into Apache Arrow RecordBatches — without deserializing, without codegen, without copying data.

pybufarrow wraps bufarrowlib, a Go library that transcodes serialized protobuf messages directly into Arrow columnar format using zero-copy memory sharing via the Arrow C Data Interface. Give it a .proto file and a stream of raw bytes; get back a pyarrow.RecordBatch you can hand straight to Pandas, Polars, DuckDB, or write to Parquet.

It's fast: the underlying engine processes ~300K messages/sec on production-shaped data — 39% faster than hand-written Arrow builders — while using 57% fewer allocations per message.

When to use pybufarrow

• Kafka / Pub-Sub consumers that receive protobuf-encoded messages and need them in columnar format for analytics, dashboards, or data lake ingestion.
• ETL pipelines where protobuf is the wire format from upstream services and the destination is Parquet, Delta Lake, or a columnar database.
• Streaming into DuckDB — denormalize protobuf messages into flat Arrow RecordBatches and append them directly to DuckDB. Aggregating pre-flattened data is orders of magnitude faster than querying nested structures.
• Ad-tech / real-time bidding — denormalize nested messages like OpenRTB BidRequests into flat, query-friendly tables with a single YAML config.
• Data science notebooks — skip the proto → dict → DataFrame dance. Go straight from wire bytes to Arrow-backed DataFrames with type fidelity.
• Feature stores / ML pipelines that ingest protobuf event streams and need low-latency materialization into Parquet or Arrow IPC.

Installation

pip install pybufarrow

Or from source (using uv):

# Build the shared library
cd bufarrowlib/cbinding && make build

# Copy into the Python package
cp cbinding/libbufarrow.so python/pybufarrow/

# Install
cd python && uv sync

Quick Start

from pybufarrow import HyperType, Transcoder

# HyperType compiles a high-performance parser from your .proto definition.
# Share one instance across all transcoders — it's thread-safe.
ht = HyperType("events.proto", "UserEvent")

with Transcoder.from_proto_file("events.proto", "UserEvent", hyper_type=ht) as tc:
    # Feed raw protobuf bytes — no deserialization needed
    for raw_bytes in kafka_consumer:
        tc.append(raw_bytes)

    # Flush to a zero-copy pyarrow RecordBatch
    batch = tc.flush()

# Use it anywhere Arrow is accepted
df = batch.to_pandas()
import duckdb
duckdb.sql("SELECT * FROM batch WHERE event_type = 'purchase'")

Proto Files with Imports

import_paths are the root directories searched for every import "..." statement inside a .proto file — and the proto_path/proto_file argument itself must also be given relative to one of those roots, the same way protoc -I <root> <relative/path.proto> works. Passing an absolute path as proto_path will fail to resolve, even if import_paths is set correctly.

Real-world example — a shared proto repository laid out one directory per package:

proto-specs/
└── docs/
    ├── OrganizationManager/
    │   └── Organization.proto      # imports google/protobuf/struct.proto
    └── Campaign/
        └── Campaign.proto          # imports OrganizationManager/Organization.proto

The shared docs/ directory is the import root; every .proto path — including cross-package imports like Campaign.proto's reference to Organization.proto — is given relative to it:

from pybufarrow import HyperType, Transcoder

import_paths = ["/path/to/platform-specs/docs"]

ht = HyperType(
    "OrganizationManager/Organization.proto",  # relative to import_paths, not absolute
    "Organization",
    import_paths=import_paths,
)

with Transcoder.from_proto_file(
    "OrganizationManager/Organization.proto",
    "Organization",
    import_paths=import_paths,
    hyper_type=ht,
) as tc:
    for raw_bytes in kafka_consumer:
        tc.append(raw_bytes)
    batch = tc.flush()

Recursive well-known types

The example above imports google/protobuf/struct.proto. Struct, Value, and ListValue are mutually recursive — Struct holds a map of Value, and Value reaches back to Struct and ListValue — so they have no finite Arrow struct representation. They arrive as a single string column holding the canonical protojson encoding, which DuckDB can query with the json_* functions:

duckdb.sql("SELECT json_extract_string(settings, '$.mode') FROM batch")

protojson output is not byte-stable across processes, so compare these columns semantically rather than by string equality, and do not content-hash them. An unset google.protobuf.Value becomes a null.

A message type that recursively contains itself is rejected when the transcoder is constructed, with an error naming the type and field path.

Streaming Batches

Process millions of messages without holding everything in memory. transcode_batch yields fixed-size RecordBatches as an iterator:

from pybufarrow import transcode_batch

# Yield 122880-row batches from a message stream
for batch in transcode_batch("events.proto", "UserEvent", message_stream, batch_size=122880):
    # Write each batch to a Parquet dataset, push to a queue, etc.
    writer.write_batch(batch)

Collect everything into a single Arrow Table, or write directly to Parquet:

from pybufarrow import transcode_to_table, transcode_to_parquet

# Arrow Table — ready for Polars, DuckDB, or any Arrow-native tool
table = transcode_to_table("events.proto", "UserEvent", messages)

# Straight to Parquet — no intermediate DataFrame
transcode_to_parquet("events.proto", "UserEvent", messages, "events.parquet")

Denormalization

Protobuf messages are often deeply nested — repeated fields, nested sub-messages, maps. Querying nested Arrow structs is painful. pybufarrow can flatten (denormalize) nested messages into wide, query-friendly rows with fan-out on repeated fields.

With a YAML config

# denorm.yaml
proto_file: order.proto
message_name: Order
denorm:
  columns:
    - name                      # top-level scalar
    - items[*].id               # fan-out: one row per item
    - items[*].price
    - seq

with Transcoder.from_config("denorm.yaml") as tc:
    for raw in order_stream:
        tc.append_denorm(raw)

    flat = tc.flush_denorm()
    # An order with 3 items produces 3 rows, each with the parent's `name` and `seq`

Programmatic column selection

ht = HyperType("order.proto", "Order")

with Transcoder.from_proto_file(
    "order.proto", "Order",
    hyper_type=ht,
    denorm_columns=["name", "items[*].id", "items[*].price", "seq"],
) as tc:
    tc.append_denorm(raw_order)
    flat = tc.flush_denorm()
    print(flat.to_pandas())
    #   name  id     price  seq
    #   acme  sku-1  9.99   1
    #   acme  sku-2  4.50   1

Empty repeated fields produce one row with nulls (left-join semantics), so you never lose parent records.

Per-message metadata columns

For source metadata such as Kafka partitions, offsets, or event timestamps, declare nullable metadata columns when constructing a denormalizer and pass values as keyword arguments. Metadata values are repeated across every fan-out row produced by that message; omitted columns become nulls:

with Transcoder.from_proto_file(
    "order.proto", "Order",
    denorm_columns=["name", "items[*].id"],
    denorm_metadata_columns=[
        ("kafka_partition", "int32"),
        ("kafka_offset", "int64"),
    ],
) as tc:
    rows = tc.append_denorm(
        raw_order,
        kafka_partition=partition,
        kafka_offset=offset,
    )
    # rows is the number of output rows produced for raw_order
    flat = tc.flush_denorm()

Use integer epoch milliseconds or microseconds for timestamp metadata, matching the declared timestamp unit. Metadata columns are configured through from_proto_file; YAML configuration does not declare them. Pool.submit and Pool.submit_merged accept the same metadata keyword arguments but remain asynchronous and do not return a row count.

Why denormalize?

Nested protobuf structures are great for wire transport but terrible for analytics. Querying nested Arrow structs or repeated fields requires unnesting at query time — which is expensive and makes aggregations slow. The denormalizer does the fan-out once at ingest time, producing flat columns that DuckDB, Pandas, and Polars can aggregate at full speed:

import duckdb

con = duckdb.connect("pipeline.duckdb")

# Create table from the first denormalized batch (schema inferred from Arrow)
con.execute("CREATE TABLE IF NOT EXISTS events AS SELECT * FROM flat_batch LIMIT 0")

# Append denormalized RecordBatches — zero-copy via Arrow
con.execute("INSERT INTO events SELECT * FROM flat_batch")

# Standard SQL on simple, flat columns — orders of magnitude faster than nested
con.sql("""
    SELECT service_name, count(*) as events, avg(latency_ms)
    FROM events
    GROUP BY service_name
    ORDER BY event_tm DESC
""")

Merging Multiple Protobuf Messages

When your pipeline enriches a base message with sidecar data (e.g., a bid request plus server-side metadata), append both in a single call:

with Transcoder.from_proto_file(
    "bidrequests.proto", "BidRequest",
    custom_proto="server_meta.proto",
    custom_message="ServerMeta",
) as tc:
    tc.append_merged(bid_request_bytes, server_meta_bytes)
    batch = tc.flush()
    # The resulting schema includes columns from both messages

Parallel Processing with Clone

clone() creates an independent transcoder that shares the same compiled schema and HyperType. Use it to fan out across threads:

import concurrent.futures
from pybufarrow import HyperType, Transcoder

ht = HyperType("events.proto", "UserEvent")
base_tc = Transcoder.from_proto_file("events.proto", "UserEvent", hyper_type=ht)

def process_partition(partition):
    with base_tc.clone() as tc:
        for msg in partition:
            tc.append(msg)
        return tc.flush()

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
    batches = list(pool.map(process_partition, partitioned_messages))

Parquet I/O

Write Arrow data to Parquet and read it back — useful for materializing transcoded streams to disk:

with Transcoder.from_proto_file("events.proto", "UserEvent", hyper_type=ht) as tc:
    for msg in messages:
        tc.append(msg)

    # Write directly to Parquet (no intermediate pyarrow step)
    tc.write_parquet("events.parquet")

    # Read back as a RecordBatch, optionally selecting columns by index
    batch = tc.read_parquet("events.parquet", columns=[0, 1, 3])

Architecture

Python user code
    ↓  Pythonic API
pybufarrow (ctypes)
    ↓  C ABI via Arrow C Data Interface
libbufarrow.so (CGo shared library)
    ↓  hyperpb TDP parser (2–3× faster than generated code)
bufarrowlib (Go)
    ↓  zero-copy Arrow record batches
pyarrow

No gRPC. No protoc codegen. No serialization round-trips. The Arrow C Data Interface means RecordBatches cross the Go→Python boundary without copying memory.

API Reference

Core Classes

Transcoder Constructors

Transcoder Methods

Batch Helpers

Requirements

• Python >= 3.9
• pyarrow >= 14.0
• libbufarrow.so / libbufarrow.dylib shared library (built from Go source via make build in cbinding/)

License

Apache-2.0