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:

Text

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.

The Kafka pipeline is configured through StreamRunConfig 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. --output-mode MODE and StreamRunConfig.output_mode accept rows (one JSON record per decision), grouped (counts per decision label), or none (no produced decision records).
• 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 produces Arrow RecordBatch objects for direct evaluation with evaluate_batch.

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

# Stream all batches from an IPC file without combining chunks.
decoder.decode_file_each(
    "events.arrow",
    lambda batch: engine.evaluate_batch(batch) or True,
)

# Requires BLAZERULES_IO_AVRO=ON. decode_batch([...]) decodes one record per
# One frame per Kafka-style message.
# per message. For a real multi-record Avro file, see
# decode_avro_ocf_file_each below instead.
decoder = blazerules_io.AvroDecoder(schema_json)
batch = decoder.decode_batch([avro_bytes])
result = engine.evaluate_batch(batch)

# Compatibility output when an NDJSON boundary is required.
ndjson = decoder.decode_ndjson([avro_bytes])

# Requires BLAZERULES_IO_PROTOBUF=ON; needs a descriptor set. Same one-frame-
# per-record shape as Avro above -- for a real multi-message file, see
# decode_delimited_file_each below instead.
decoder = blazerules_io.ProtobufDecoder(descriptor_set_bytes, "package.Transaction")
batch = decoder.decode_batch([proto_bytes])
result = engine.evaluate_batch(batch)

# Compatibility output when an NDJSON boundary is required.
ndjson = decoder.decode_ndjson([proto_bytes])

Arrow IPC decoder methods:

ArrowIpcReadOptions controls field projection and validation. Use included_fields
to avoid decoding unused top-level fields. validation accepts full,
structural, or trusted; external untrusted frames should use full.

📘

File decoding from the CLI

Arrow IPC files are naturally multi-batch/multi-record and blazerules eval reads all of them:

Bash

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:
  Bash

  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:
  Bash

  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):
  Bash

  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)

# Callback form avoids collecting all batches before processing.
blazerules_io.for_each_record_batch(
    "history/day.parquet",
    "parquet",
    lambda batch: engine.evaluate_batch(batch) or True,
)

# 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]
for_each_record_batch(path: str, format: str, callback, options: FileReadOptions) -> int
read_ndjson_bytes(path: str) -> bytes

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

Related documentation