Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Your first project

Every release publishes the example projects beside the binary. customer-metrics is the smallest project that shows the whole idea: one source, one model, one fixture. Everything below is real output from that project.

curl -fsSL https://github.com/glyf-data/streamform-releases/releases/latest/download/streamform-examples.tar.gz | tar xz
cd streamform-examples/customer-metrics

Three files

The source: a Kafka topic, so the input never ends. The schema names the columns and their types.

# sources.yml
sources:
  orders:
    connector: kafka
    topic: orders
    format: json
    schema:
      customer_id: int64
      amount: decimal

The model: plain SQL. source('orders') reads the source declared above.

-- models/customer_metrics.sql
SELECT
    customer_id,
    SUM(amount) AS total_spend
FROM source('orders')
GROUP BY customer_id

The project file: the model’s result is keyed by customer_id and materialized as an upsert. Streamform checks that this agrees with what the SQL produces; a GROUP BY without upsert is an error, not a guess.

# streamform.yml
name: customer-metrics
version: 1

models:
  customer_metrics:
    materialized: upsert
    key:
      - customer_id

Check it

streamform check

customer-metrics (schema version 1)

models
  customer_metrics         upsert(customer_id)      models/customer_metrics.sql

sources
  orders                   kafka/json   topic=orders             2 columns

fixtures
  customer_metrics         tests/customer_metrics.yml 2 given, 2 expect

ok: 1 model, 1 source, 1 fixture, 0 errors, 0 warnings

check compiles every model to its streaming plan and reports anything that does not plan: an unsupported construct, a ref() to a model that does not exist, a declared key that disagrees with the GROUP BY, a fixture that names a column the model does not output.

Read what the SQL means

streamform explain customer_metrics

customer_metrics

Source
  name=orders
  connector=kafka
  time=processing
  boundedness=Unbounded
  changelog=Append
  state=Stateless
     ↓
Aggregate
  key=customer_id
  aggregates=sum(amount)
  boundedness=Unbounded
  changelog=Upsert(customer_id)
  state=Keyed
  fingerprint=aggregate key=[customer_id] accumulators=[sum(amount): decimal(38,9)] retention=unbounded
  digest=4aead023aa01
     ↓
Project
  columns=customer_id, sum(amount) AS total_spend
  boundedness=Unbounded
  changelog=Upsert(customer_id)
  state=Stateless
     ↓
Sink
  model=customer_metrics
  materialized=upsert
  boundedness=Unbounded
  changelog=Upsert(customer_id)
  state=Stateless

Read it top to bottom: the source never ends, grouping it needs keyed state (one accumulator per customer, a decimal(38,9) running total), and from the aggregate onward the result is an updating stream keyed by customer_id. The concepts section explains each line.

Test it

The fixture gives the model two events and states the exact changelog it must emit.

# tests/customer_metrics.yml
model: customer_metrics

given:
  - customer_id: 42
    amount: 10
  - customer_id: 42
    amount: 18

expect:
  - op: insert
    customer_id: 42
    total_spend: 10
  - op: update
    customer_id: 42
    total_spend: 28

streamform test

customer_metrics

PASS

2 events processed
2 changelog records emitted

ok: 1 fixture passed

That ran through the in-process simulator: no Kafka, no cluster, no wall clock, and the same result every time. Change the second expectation to total_spend: 27 and the failure shows expected and actual side by side, with the first difference named:

  #  expected                              actual
  1  insert customer_id=42 total_spend=10  insert customer_id=42 total_spend=10
  2  update customer_id=42 total_spend=27  update customer_id=42 total_spend=28
     ^ total_spend: expected 27, got 28

Next