Recipe: Binary Format Ingestion

Binary decoders produce Arrow RecordBatch objects and feed evaluate_batch. They do not convert through JSON.

Arrow IPC

import blazerules
import blazerules_io

engine = blazerules.RuleEngine()
engine.load_rules("rules.yaml")

decoder = blazerules_io.ArrowIpcDecoder()
batch = decoder.decode_batch([frame_bytes])
result = engine.evaluate_batch(batch)

blazerules eval --rules rules.yaml --input arrow-ipc --path batch.arrow --output summary

Streaming large files (Arrow IPC, Parquet, CSV)

decode_batch above (and its list-returning sibling decode_batches) are one-shot: pass them frames you already hold in memory (a Kafka message, an HTTP body) and get Arrow batches back. For files on disk or s3:// — where the whole object may not fit comfortably in memory — for_each_record_batch streams one RecordBatch at a time instead, and supports early exit:

import blazerules_io

def handle_batch(batch):
    engine.evaluate_batch(batch)
    return True  # return False to stop reading early

blazerules_io.for_each_record_batch("events.arrow", "arrow-ipc", handle_batch)
blazerules_io.for_each_record_batch("history.parquet", "parquet", handle_batch)
blazerules_io.for_each_record_batch("events.csv", "csv", handle_batch)

Pass FileReadOptions.included_fields to push column selection into the reader instead of dropping columns after the fact — Parquet skips those row-group columns entirely and CSV's converter never parses them:

options = blazerules_io.FileReadOptions()
options.included_fields = ["event_id", "amount", "merchant_id"]
blazerules_io.for_each_record_batch("history.parquet", "parquet", handle_batch, options)

📘

Column projection is per-format

included_fields (field names) is honored by the Parquet and CSV readers. Arrow IPC projects by included_field_indices (integer positions) instead — set that field on the same FileReadOptions if you need to project an .arrow/.feather file. NDJSON reads currently ignore both and always return every field.

read_record_batches(path, format, batch_size) still exists and is simpler for small files — it calls the same reader internally but collects every batch into a list before returning, so it does not accept a FileReadOptions:

for batch in blazerules_io.read_record_batches("history.parquet", batch_size=65536):
    engine.evaluate_batch(batch)

Prefer for_each_record_batch once a file is large enough that materializing every batch at once would be a meaningful chunk of memory — blazerules eval uses the same streaming form internally.

Avro

AvroDecoder.decode_batch([...]) decodes one record per bare frame you pass in — the right shape for a Kafka-style topic (one frame per message):

decoder = blazerules_io.AvroDecoder(schema_json)
batch = decoder.decode_batch([avro_bytes])
result = engine.evaluate_batch(batch)

For a real multi-record Avro file — an Object Container File (OCF), the format Spark/Hadoop/Kafka Connect produce — use decode_avro_ocf_file_each instead. It reads the file's own embedded schema (no schema_json needed) and streams batches:

def handle_batch(batch):
    engine.evaluate_batch(batch)
    return True  # return False to stop reading early

blazerules_io.decode_avro_ocf_file_each("events.avro", handle_batch, batch_size=10000)

# Object Container File: every record, auto-detected via magic bytes, no --schema needed
blazerules eval --rules rules.yaml --input avro --path events.avro --output summary

# Bare single Avro value (e.g. one Kafka message payload saved to disk): needs --schema
blazerules eval --rules rules.yaml --input avro --schema transaction.avsc --path record.avrobin --output summary

Protobuf

ProtobufDecoder.decode_batch([...]) decodes one record per bare frame, the same one-frame-per-record shape as Avro above:

decoder = blazerules_io.ProtobufDecoder(descriptor_set_bytes, "package.Transaction")
batch = decoder.decode_batch([proto_bytes])
result = engine.evaluate_batch(batch)

Protobuf has no self-describing container format and no magic bytes, so a real multi-message file needs an explicit convention: N varint-length-delimited messages (the same convention protobuf's own SerializeDelimitedToCodedStream/ParseDelimitedFromCodedStream use). Read one with decode_delimited_file_each, or decode_delimited_file_parallel to parse chunks concurrently across worker threads once a file is large enough for that to matter:

def handle_batch(batch):
    engine.evaluate_batch(batch)
    return True  # return False to stop reading early

blazerules_io.ProtobufDecoder(descriptor_set_bytes, "package.Transaction") \
    .decode_delimited_file_each("events.pb", handle_batch, batch_size=10000)

# Bare single message: decodes exactly one record, unchanged
blazerules eval --rules rules.yaml --input protobuf --descriptor descriptor.pb --message package.Transaction --path transaction.pb --output summary

# N varint-length-delimited messages: decodes every record, not auto-detected
blazerules eval --rules rules.yaml --input protobuf-delimited --descriptor descriptor.pb --message package.Transaction --path events.pb --output summary

Nested Data

Nested structs use dotted rule fields:

conditions:
  field: merchant.risk.score
  op: gt
  value: 50

Arrays of objects use array_any:

conditions:
  array_any:
    path: items
    where:
      and:
        - field: price
          op: gt
          value: 100
        - field: category
          op: eq
          value: electronics