Stream Processing for the 98%

Higgins is a simple streaming engine designed to give you everything you need, whilst not forcing you to deploy and maintain a cluster.

Get started View on GitHub

Schema

for keeping your data in check.

Functions

for data transformations.

Configurable Storage

for decoupled compute

Push or poll based API

for atleast-once delivery.

Getting Started

Configure

A stream is a schema'd and partitioned log. Example events stream keyed by id:

config.toml

[storage.memory]
type = "memory"

[schema.event]
id      = "string"
message = "string"

[streams.events]
schema        = "event"
partition_key = "id"

Upload Configuration

terminal

# Start the broker
$ ./higgins

main.rs

client.upload_configuration(config.as_bytes())?;

let upload_config_response = client.recv(None).await?;

Produce a record

main.rs

let schema = Arc::new(Schema::new(vec![
    Field::new("id",      DataType::Utf8, false),
    Field::new("message", DataType::Utf8, false),
]));

client
    .produce_json("events", br#"{"id":"1","message":"hello"}"#, schema)
    .await?;

client.recv(None).await?;

Read it back

main.rs

client
    .query_at(b"events", &PartitionName::try_from("1")?, 0)
    .await?;

if let ResponseBody::GetIndex(resp) = client.recv(None).await?.body {
    if let Some(batch) = read_arrow(&resp.records[0].data)?.next()? {
        println!("{batch:?}");
    }
}

Configurations to Topographies

Streams, schema and storage are defined in configurations and uploaded to the broker. Configurations get folded into the broker's topography.

Typed schema. Every field maps to an Arrow type — strings, ints, floats, dates and more.
Keyed. Records shard by a key
Configurable storage. Object storage initially, sky is the limit.

config.toml

[schema.amount]
id   = "string"
data = "int32"

[streams.amount]
schema        = "amount"
partition_key = "id"

[streams.result] 
base          = "amount"
type          = "reduce"
partition_key = "id"
schema        = "amount"
fn            = "sum"

Things to do with streams

Build map, reduce,join and window, and higgins keeps it continuously up to date.

Mappings

1:1. A user defined functions transforms every record.

type = "map"
base = "amount"
fn   = "double"   # UDF that doubles the amount

Reduce

Stateful accumulation. Each record folds into the previous result.

# inputs   1, 2, 3
# results  1, 3, 6
type = "reduce"  fn = "sum"

Joins

Join datasets through their matching keys, the latest of each side is carried forward

type = "join"
base = "customer"
join = { type = "inner", stream = "address" }

Windowing

Aggregate the most recent N records per key.

type   = "window"
window = { type = "count", interval = "5" }