Skip to main content
krabka
GitHub
Go 1.26+ · Track v0.2.0

Go Streams & Arrow

Stream processing in Go with Apache Arrow record batches, barrier cuts, and schema decoding via krabka-streams-go.

Browse Godoc Reference

krabka-streams-go processes Kafka records as Apache Arrow batches. A consumer poll is decoded into a record batch, passed through batch operators, and encoded back into Kafka records at a sink.

Architecture & Engineering Guides

In-depth technical guides maintained alongside the Go codebase:

  • Getting Started: Installation, topology builder walkthrough, and consumer group setup.
  • Configuration: WithDefaults and the settings it applies.
  • Columnar Processing: Arrow batch filters, memory allocators, and record projection.
  • Serdes: Avro, Protobuf, and JSON schema caching and binary decoding.
  • Schema Registry: Confluent Schema Registry client integration, schema evolution, and wire protocol.
  • Barrier Cuts: Snapshot alignment, state checkpoints, and non-blocking barriers.
  • Coordination Primitives: Leader election, leases, and fencing tokens.
  • Testing: In-memory pipeline testing and deterministic record assertions.
  • Architecture: Topology construction, channel-based execution graph, and memory safety.

Installation

Go Module Command 
go get github.com/krabka-io/krabka-streams-go

Packages

Package Purpose
columnarApache Arrow batch processing, topologies, and barrier cuts
schemaRegistry client, cache, and Avro/Protobuf/JSON Schema serdes
coordinationLeader election, leases, and fencing tokens
columnarschemaAvro and Protobuf Arrow memory bridges

Columnar Processing Example

Topology Definition 
package main

import (
	"log"

	"github.com/apache/arrow-go/v18/arrow"
	"github.com/apache/arrow-go/v18/arrow/array"
	"github.com/apache/arrow-go/v18/arrow/memory"

	"github.com/krabka-io/krabka-streams-go/columnar"
)

func main() {
	mem := memory.NewGoAllocator()
	codec := columnar.NewBlobCodec(mem)
	topology := columnar.NewTopology(mem)

	source, err := topology.AddSource("source", []string{"transactions"}, codec)
	if err != nil {
		log.Fatal(err)
	}
	// Records carry Arrow IPC streams; column 1 of the payload is an int64 amount.
	large, err := topology.AddOperator("large",
		columnar.Filter(mem, func(batch arrow.Record, row int) bool {
			return batch.Column(1).(*array.Int64).Value(row) > 4
		}), source)
	if err != nil {
		log.Fatal(err)
	}
	if _, err := topology.AddSink("archive", "large-transactions", codec, large); err != nil {
		log.Fatal(err)
	}

	built, err := topology.Build()
	if err != nil {
		log.Fatal(err)
	}
	_ = built
}

Version & Compatibility

Component Supported Version Notes
Go Runtime Go 1.26+ go.mod declares go 1.26.5
Apache Arrow Go arrow-go v18.7.0 Native Go allocator and Record batches
krabka Broker Apache Kafka 4.x or a krabka broker Wire-protocol compatibility

Full version matrices and SLSA Level 3 build provenance records are tracked on the Versions page.