Streaming & IO

blazerules_io adds streaming sources/sinks and binary decoders. The release wheel and default source build include it, while custom lean builds can still disable it.

📘
Check capabilities at runtime
Full builds enable IO, Kafka, Avro, and Protobuf. Custom lean builds can turn them off, so check availability before wiring production code:
BLAZERULES_IO=ON
BLAZERULES_IO_KAFKA=ON
BLAZERULES_IO_AVRO=ON
BLAZERULES_IO_PROTOBUF=ON
See Installation and Configuration Reference.

What it provides

Kafka source/sink through librdkafka.
Debezium CDC unwrap.
Arrow IPC frames.
Avro binary records.
Protobuf binary records with descriptor sets.
Local and exact-object s3:// file reads.

Binary decoders produce Arrow RecordBatch objects and call evaluate_batch directly — they do not convert through JSON.

Capability flags

Connectors and decoders depend on build flags, so check availability at runtime before using them:

import blazerules_io

print(blazerules_io.has_kafka)
print(blazerules_io.has_avro)
print(blazerules_io.has_protobuf)

Kafka

Consume a batch, evaluate it, and produce decisions. See the full walkthrough in Kafka Streaming.

import blazerules, blazerules_io

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

consumer = blazerules_io.KafkaConsumer(
    brokers="localhost:9092",
    group_id="blazerules-workers",
    topics=["transactions"],
    conf={"enable.auto.commit": "false"},
)
producer = blazerules_io.KafkaProducer(
    brokers="localhost:9092",
    conf={},
)

The constructor signatures are:

KafkaConsumer(brokers: str, group_id: str, topics: list[str], conf: dict[str, str] = {})
KafkaProducer(brokers: str, conf: dict[str, str] = {})

For a fully C++-driven loop, the module also exposes run_stream(engine, config). It can consume JSON/NDJSON, Debezium envelopes, Arrow IPC frames, Avro frames, or Protobuf frames depending on payload_format.

cfg = blazerules_io.StreamRunConfig()
cfg.brokers = "localhost:9092"
cfg.group_id = "blazerules-workers"
cfg.input_topics = ["transactions"]
cfg.output_topic = "decisions"
cfg.batch_size = 8192
cfg.poll_timeout_ms = 200
cfg.flush_timeout_ms = 5000
cfg.max_messages = 0
cfg.max_batches = 0
cfg.commit_offsets = True
cfg.payload_format = "json"       # json, ndjson, debezium, arrow-ipc, avro, protobuf
cfg.avro_schema_json = ""          # required for avro
cfg.protobuf_descriptor_set = b""   # required for protobuf
cfg.protobuf_message_type = ""      # required for protobuf

stats = blazerules_io.run_stream(engine, cfg)
print(stats.batches, stats.messages, stats.matched, stats.emitted, stats.eval_us)

The equivalent native CLI form is:

blazerules stream kafka \
  --rules rules.yaml \
  --brokers localhost:9092 \
  --input-topic transactions \
  --output-topic decisions \
  --format protobuf \
  --descriptor schema.desc \
  --message payments.Transaction

Kafka pipeline architecture

run_stream and blazerules stream kafka run a concurrent pipeline rather than a single-threaded consume → evaluate → produce → commit loop: one main thread polls Kafka and dispatches messages by partition hash to per-worker bounded queues, worker_count worker threads evaluate independently (each against its own RuleEngine shard from engine.create_shards, so no Kafka calls happen on these threads — pure CPU work), and one delivery thread performs all Kafka produce/flush calls and commits offsets only after a successful flush.

New in 0.5.3, the pipeline is tuned with these StreamRunConfig fields and matching blazerules stream kafka flags:

Worker threads. Set --workers N (CLI) or StreamRunConfig.worker_count (Python) to the number of concurrent evaluation worker threads, each with its own engine shard (default 1). Pure CPU work — no Kafka calls happen on these threads.
Queue depth. Set --queue-depth N (CLI) or StreamRunConfig.queue_depth (Python) to size every bounded queue in the pipeline — each worker's input queue and the shared delivery queue all use this one value (default 64).
Flush interval. Set --flush-interval-ms N (CLI) or StreamRunConfig.flush_interval_ms (Python) to control how often the delivery thread batches and flushes produced messages to Kafka (default 250). Offsets commit only after a successful flush — the consumer now sets enable.auto.offset.store=false explicitly, where earlier versions relied on the default auto-store, which only stayed safe because the old loop was strictly serial.
Partition affinity. Set --partition-affine (CLI) or StreamRunConfig.partition_affine (Python, default true) to route every message for a given (topic, partition) to the same worker, preserving per-partition order. This is forced to true whenever commit_offsets is enabled, since ordering is required for correct offset commit semantics; it is only optional (round-robin dispatch across workers) when commit_offsets=false.
Output mode. Set --output-mode MODE (CLI) or StreamRunConfig.output_mode (Python) to rows (default; one JSON record per decision, the previous behavior), grouped (per-decision-label counts instead of one row per record — a throughput win when you only need aggregate stats), or none (disable output entirely).
Arrow validation level. Set --arrow-validation LEVEL (CLI) or StreamRunConfig.arrow_validation (Python) to full, structural (default), or trusted — the validation level applied to Arrow-IPC-encoded Kafka payloads before decoding. See Binary decoders for the decoder itself.

📘
Kafka payloads decode zero-copy
Arrow IPC frames consumed from Kafka are decoded directly out of librdkafka's message buffer (aliased through a shared_ptr owner) instead of being deep-copied per message — a memory/CPU efficiency change with no effect on decoded values.

🚧
A crash can now replay a larger batch
Delivery guarantees are unchanged — still at-least-once, produce-before-commit — but commits now happen on the flush_interval_ms timer across many concurrently evaluated messages instead of synchronously after every single one. A crash can therefore leave a larger batch of already-processed messages to be reprocessed on restart than under the old strictly-serial loop. This is a throughput vs. replay-window trade-off: lower flush_interval_ms for a tighter bound on reprocessing, at some cost to throughput.

Debezium CDC

unwrap_debezium turns Debezium change events into evaluable NDJSON (the op field defaults to __op).

ndjson = blazerules_io.unwrap_debezium(messages, op_field="__op")
result = engine.evaluate_ndjson(ndjson)

unwrap_debezium(...) returns one contiguous NDJSON bytes object. Pass that directly to RuleEngine.evaluate_ndjson(...).

Binary decoders

Each decoder turns binary frames into an Arrow RecordBatch you pass straight to evaluate_batch.

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

📘
File decoding from the CLI
Arrow IPC files are naturally multi-batch/multi-record and blazerules eval reads all of them:
blazerules eval --rules rules.yaml --input arrow-ipc --path events.arrow
Avro and Protobuf frames have no inherent file-level framing (a bare Avro/Protobuf
value doesn't say where it ends), so a file needs one of two shapes:
Avro Object Container File (OCF) -- the format Spark/Hadoop/Kafka Connect
produce, with an embedded schema, sync markers, and one or more blocks of
records. --input avro auto-detects this (via its magic bytes) and decodes
every record in the file; --schema isn't needed since the file carries its
own:
blazerules eval --rules rules.yaml --input avro --path events.avro
A file that isn't OCF-framed is treated as one bare Avro-encoded value (e.g.
a single Kafka message payload saved to disk) and decodes to exactly one
record -- this path still needs --schema:
blazerules eval --rules rules.yaml --input avro --path one_message.avro --schema events.avsc
Protobuf has no OCF equivalent and no magic bytes, so multi-message files
need an explicit opt-in rather than auto-detection (a delimited file is
indistinguishable from a single bare message by content alone). Use
--input protobuf-delimited for a file of N varint-length-prefixed messages
(the same convention protobuf's own SerializeDelimitedToCodedStream/
ParseDelimitedFromCodedStream use):
blazerules eval --rules rules.yaml --input protobuf-delimited --path events.pb \
  --descriptor schema.desc --message payments.Transaction
Plain --input protobuf --path one_message.pb ... still decodes exactly one
bare message, unchanged, so existing single-message files keep working.
The Python-side equivalents (blazerules_io.decode_avro_ocf_file_each,
ProtobufDecoder.decode_delimited_file_each/decode_delimited_file_parallel)
give the same capability without shelling out to the CLI.

File readers

Read local or s3:// files into Arrow batches or NDJSON bytes.

# Iterate Arrow RecordBatches from Parquet/Arrow/CSV:
for batch in blazerules_io.read_record_batches("history/day.parquet", batch_size=65536):
    result = engine.evaluate_batch(batch)

# Read an NDJSON file (local or s3://) as bytes:
payload = blazerules_io.read_ndjson_bytes("s3://bucket/events/day.ndjson")
result = engine.evaluate_ndjson(payload)

Signatures:

read_record_batches(path: str, format: str = "auto", batch_size: int = 65536) -> list[pyarrow.RecordBatch]
read_ndjson_bytes(path: str) -> bytes

See S3 resources for AWS profile/region/endpoint configuration.

Where to go next

Kafka Streaming

End-to-end consume → evaluate → produce tutorial.

CLI Agent

A configured multi-instance ingest agent — no custom loop required.