Stream Processing

Flo ships a built-in stream processing engine inspired by Apache Flink. Jobs are defined in YAML, submitted via the CLI or SDKs, and executed entirely inside the Flo node — no external cluster required. The engine reads from Flo streams or time-series measurements, transforms records through a chain of operators, and writes results to stream, time-series, KV, or queue sinks.

Why Built-In Processing?

Traditional architectures pair a storage layer (Kafka, Redis) with a separate processing framework (Flink, Spark Streaming). This means two clusters to deploy, two failure domains to monitor, and serialization overhead at every boundary.

Flo eliminates that gap:

• Zero-copy reads — source records are read directly from the Unified Append Log, no network hop
• Thread-per-shard execution — each pipeline runs on the shard that owns the source partition, no cross-thread coordination
• Integrated state — operators can read and write Flo KV directly (e.g., kv_lookup), no external state store
• Single deployment — one binary, one config, one monitoring surface

Core Concepts

Jobs

A job is a long-running processing pipeline defined by a YAML file. Each job specifies:

Records

The fundamental data unit flowing through a pipeline is a ProcessingRecord:

• key — optional partitioning key (empty = broadcast)
• value — payload bytes (typically JSON)
• event_time_ms — event timestamp used for windowing and watermarks
• source — origin stream name, partition, and offset for checkpoint tracking
• headers — optional key-value metadata

Pipeline Topology

Pipelines are linear chains: source → operator₁ → operator₂ → … → sink. All operators in a chain execute on the same shard thread with zero-copy record passing (fused execution, mirroring Flink's operator chaining).

┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
│  Source   │───▶│  Filter  │───▶│  KeyBy   │───▶│ Aggregate│───▶│   Sink   │
│ (stream)  │    │          │    │          │    │          │    │ (stream) │
└──────────┘    └──────────┘    └──────────┘    └──────────┘    └──────────┘

Job Definition (YAML)

Server-side job definitions use nested YAML or JSON. The examples below use YAML.

Minimal Pipeline

The simplest pipeline: read a stream, write to another stream, no transformation:

kind: Processing
name: user-events-pipe
namespace: analytics
parallelism: 1
batch_size: 100
 
sources:
  - name: input
    stream:
      name: payment-events
      namespace: analytics
 
sinks:
  - name: output
    stream:
      name: enriched-events
      namespace: analytics

The job polls payment-events on a fixed interval (poll_interval_ms, default 1000 ms), reading up to batch_size records per tick, and appends every record to enriched-events.

Full Example with Operators

kind: Processing
name: order-analytics
namespace: production
parallelism: 4
batch_size: 200
 
sources:
  - name: orders
    stream:
      name: raw-orders
      namespace: production
      batch_size: 500
 
sinks:
  - name: output
    stream:
      name: order-totals
      namespace: production
 
operators:
  - type: filter
    name: completed-only
    condition: "value_contains:completed"
 
  - type: keyby
    name: by-customer
    key_expression: "$.customer_id"
 
  - type: aggregate
    name: hourly-total
    function: sum
    field: "$.amount"
    window: tumbling
    window_size: 3600

Sources

Sources define where the pipeline reads data. Two source kinds are supported: stream and time-series (ts).

Stream Source

Reads records from a Flo stream, tracking a durable offset cursor for at-least-once processing (a crash redelivers at most the last debounce window — never skips records):

sources:
  - name: events
    stream:
      name: click-events
      namespace: production
      partitions: 0           # single partition (integer)
      batch_size: 500         # per-source override (records read per tick)
      poll_interval_ms: 1000  # how often to read from the source (default: 1000)

Partition Selection

Time-Series Source

Polls a TS measurement at a configurable interval, emitting data points as JSON records:

sources:
  - name: cpu-metrics
    ts:
      measurement: cpu_usage
      namespace: production
      field: value              # field to read (default: "value")
      tags:                     # optional tag filter
        host: web-01
      poll_interval_ms: 1000    # polling interval (default: 1000)

Each emitted record looks like:

{"measurement": "cpu_usage", "value": 72.5, "timestamp_ms": 1700000000000}

Multi-Source Merge

Multiple sources are polled round-robin into the same pipeline:

sources:
  - name: clicks
    stream:
      name: click-events
  - name: checkouts
    stream:
      name: checkout-events
 
sinks:
  - name: merged
    stream:
      name: unified-events

Sinks

Sinks define where the pipeline writes results. Four sink kinds are supported.

Stream Sink

Appends records to a Flo stream:

sinks:
  - name: output
    stream:
      name: enriched-events
      namespace: production

Time-Series Sink

Extracts tags and numeric fields from JSON record payloads and writes them as time-series data points:

sinks:
  - name: cpu-ts
    ts:
      measurement: proc_cpu_metrics
      namespace: metrics
      tags:
        host: hostname          # TS tag "host" ← JSON field "hostname"
      fields:
        cpu: cpu_percent        # TS field "cpu" ← JSON field "cpu_percent"

Given the input {"hostname":"web-01","cpu_percent":72.5}, the sink writes a point to proc_cpu_metrics with tag host=web-01 and field cpu=72.5.

value_field Shorthand

If the JSON payload contains a single scalar field to use as the value, use value_field instead of a full fields: map:

sinks:
  - name: temp
    ts:
      measurement: proc_temp
      value_field: temperature
      tags:
        sensor: sensor

KV Sink

Writes records to Flo KV storage:

sinks:
  - name: profiles
    kv:
      namespace: profiles
      key_prefix: user          # prepended to record key
      separator: ":"            # between prefix and key (default: ":")
      write_mode: upsert        # upsert | if_absent | versioned
      ttl_ms: 86400000          # optional TTL

Queue Sink

Enqueues records into a Flo queue:

sinks:
  - name: tasks
    queue:
      name: task-queue
      namespace: default
      priority: 5
      delay_ms: 1000            # visibility delay

Tag-Based Routing

Sinks can subscribe to tagged records instead of receiving the full pipeline output. Tags are set by the classify operator and matched with AND logic — a record reaches a sink only when it carries all of the sink's declared match tags.

Up to 32 distinct tag names per pipeline.

operators:
  - type: classify
    name: route-events
    rules:
      - condition: "json:level=error"
        tag: errors
      - condition: "json:$.latency_ms>5000"
        tag: slow
 
sinks:
  - name: main
    stream:
      name: results
 
  - name: late-events
    stream:
      name: late-data
    match: [late]
 
  - name: failures
    queue:
      name: dead-letter
    match: [errors]
 
  - name: slow-errors
    stream:
      name: slow-error-stream
    match: [errors, slow]          # AND — must have BOTH tags

Sinks without match receive all records from the main operator chain. A single match: value is also accepted as sugar: match: late is equivalent to match: [late].

Operators

Operators are the transformation steps between source and sink. They are declared in the operators: array and execute in order. Flo includes eight built-in operator types.

filter — Expression-Based Filter

Drops records that don't match a condition. Supported expressions:

Record-level:

JSON field — json:<path><op><value>:

operators:
  - type: filter
    name: keep-important
    condition: "value_contains:important"

keyby — JSON Key Extraction

Re-keys records by extracting a field from the JSON value using JSONPath expressions. Downstream operators like aggregate group by this key.

operators:
  - type: keyby
    name: by-user
    key_expression: "$.user_id"

Supports nested paths: "$.metadata.region".

If the value isn't valid JSON or the field doesn't exist, the record passes through with its original key.

map — Declarative Field Projection

A 1:1 transformation that restructures JSON records by projecting, renaming, and adding fields:

operators:
  - type: map
    name: extract-user
    user_id: "$.data.user_id"        # JSONPath extraction
    amount: "$.transaction.amount"    # nested field
    source: "payment-service"         # constant string

Rules:

• Values starting with $. are JSONPath extractions from the input
• All other values are treated as string constants
• Output is always a JSON object with the declared fields

flatmap — Array Explosion

A 1:N transformation that explodes a JSON array into individual records:

operators:
  - type: flatmap
    name: explode-items
    array_field: "$.order.items"
    element_key: "$.sku"            # optional: extract key from each element

Each element of the array at $.order.items becomes a separate downstream record. If the field is missing or not an array, the record is dropped.

aggregate — Declarative Aggregation

Accumulates numeric values extracted from JSON records. Records are grouped by key — chain a keyby operator before aggregate for custom grouping.

Functions: sum, count, avg, min, max

Window modes:

Output format: {"key": <group-key>, "value": <result>, "count": <n>} (the key is the grouping key set by a preceding keyby, or the record's key otherwise)

operators:
  # Running total (emits on every record)
  - type: aggregate
    name: running-total
    function: sum
    field: "$.amount"
 
  # Count per key in 60-second tumbling windows
  - type: aggregate
    name: event-counter
    function: count
    window: tumbling
    window_size: 60
 
  # Average every 100 records per key
  - type: aggregate
    name: batch-avg
    function: avg
    field: "$.latency_ms"
    window: count
    window_size: 100

kv_lookup — KV-Backed Enrichment & Filtering

Looks up keys in Flo KV storage to filter or enrich streaming records. The lookup key is constructed from a template that interpolates JSON fields from the record.

Template syntax: Mix literal text with ${<jsonpath>} placeholders:

Modes:

operators:
  - type: kv_lookup
    name: check-account
    lookup_key: "account:${$.account_id}"
    namespace: default
    mode: filter

Enrich example — adds a "tier" field to each record:

operators:
  - type: kv_lookup
    name: enrich-user
    lookup_key: "user:${$.user_id}"
    namespace: default
    mode: enrich
    enrich_field: tier

classify — Tag-Based Record Labeling

Evaluates rules against each record and sets tag bits for matching conditions. Records continue flowing — classify never drops or forks, it only labels. Downstream sinks filter by AND-matching their tags list against the record's tag bitfield.

Rules use the same condition expressions as filter.

operators:
  - type: classify
    name: route-payments
    default_tag: unmatched           # optional: tag when no rules match
    rules:
      - condition: "json:type^=payment"
        tag: payments
      - condition: "json:type*=transfer"
        tag: transfers
      - condition: "json:amount>10000"
        tag: high-value

A single record can match multiple rules and carry multiple tags. This enables multi-way routing: a payments sink, a high-value sink, and a high-value, payments sink can all coexist.

Records that match no rules pass through untagged by default. Set default_tag to route unmatched records to a specific sink.

passthrough — Identity / Debug Tap

Emits every record unchanged. Useful for testing and as a pipeline placeholder:

operators:
  - type: passthrough
    name: debug-tap

Chaining Operators

Operators execute in declaration order. Records flow through each operator sequentially:

operators:
  - type: filter
    name: completed-only
    condition: "value_contains:completed"
 
  - type: map
    name: extract-fields
    order_id: "$.order.id"
    total: "$.order.total"
 
  - type: keyby
    name: by-customer
    key_expression: "$.customer_id"
 
  - type: aggregate
    name: daily-total
    function: sum
    field: "$.total"
    window: tumbling
    window_size: 86400

Windowing

The processing engine supports event-time windowing for time-based aggregations.

Window Types

Triggers

Triggers decide when a window fires (evaluates its function and emits results):

Watermarks

Watermarks signal event-time progress through the pipeline. They propagate the guarantee that no record with an earlier timestamp will arrive. The engine tracks watermarks across multiple inputs and advances them based on the minimum across all sources.

Late Data Handling

Records that arrive after the watermark has passed their window are classified as:

1. On-time — within the expected window
2. Late — past the watermark but within the allowed lateness (re-fires window)
3. Dropped — too late, beyond the allowed lateness threshold

Late records can be tagged (via a classify rule) and routed to a dedicated sink for monitoring or reprocessing.

Keyed State

Stateful operators (like aggregate) access per-key state scoped to the job and operator instance. Three state primitives are available:

State is stored in the per-shard state backend with automatic key namespacing:

proc:{job_id}:{operator_id}:kv:{user_key}          — ValueState
proc:{job_id}:{operator_id}:list:{user_key}:{idx}  — ListState
proc:{job_id}:{operator_id}:map:{user_key}:{mk}    — MapState

No locks are needed — single-threaded shard execution guarantees isolation.

Checkpoints and Savepoints

Checkpoints

The engine implements Chandy-Lamport distributed snapshots for fault tolerance:

1. Trigger — coordinator assigns a new checkpoint_id
2. Barrier injection — CheckpointBarrier is injected into sources
3. Source snapshot — sources save their current offsets
4. Operator snapshot — each operator snapshots its state, forwards the barrier
5. Sink acknowledgment — sinks confirm receipt of the barrier
6. Completion — all acks received, checkpoint is marked complete

Checkpoint metadata is persisted internally under _proc: keys in KV storage.

Savepoints

Savepoints are user-triggered snapshots of a job's state:

flo processing savepoint job-1 -n analytics

A savepoint captures:

• The job binding and configuration
• Creation timestamp
• records_processed at the savepoint time

You can restore a job from a savepoint:

flo processing restore job-1 sp-1 -n analytics

Pipeline Patterns

Stream to Stream (Passthrough)

The simplest pipeline — data flows unmodified from source to sink:

kind: Processing
name: mirror
sources:
  - stream: { name: events }
sinks:
  - stream: { name: events-copy }

Stream to Time-Series (Metrics Extraction)

Convert JSON event streams into queryable time-series data:

kind: Processing
name: cpu-pipeline
namespace: metrics
sources:
  - stream: { name: cpu-json }
sinks:
  - ts:
      measurement: proc_cpu_metrics
      tags:
        host: hostname
      fields:
        cpu: cpu_percent

Time-Series to Stream (Alerting Feed)

Poll TS measurements and produce JSON records for downstream consumers:

kind: Processing
name: ts-to-alerts
namespace: metrics
sources:
  - ts:
      measurement: cpu_usage
      field: value
      tags:
        host: web-01
sinks:
  - stream: { name: cpu-alerts }

Time-Series to Time-Series (Derived Metrics)

Build derived metrics from raw measurements:

kind: Processing
name: derive-temperature
namespace: metrics
sources:
  - ts:
      measurement: raw_temperature
      field: value
sinks:
  - ts:
      measurement: derived_temp
      value_field: value

KV-Enriched Pipeline

Filter or enrich events using data stored in Flo KV:

kind: Processing
name: account-filter
namespace: production
sources:
  - stream: { name: transactions }
sinks:
  - stream: { name: valid-transactions }
operators:
  - type: kv_lookup
    name: check-account
    lookup_key: "account:${$.account_id}"
    namespace: production
    mode: filter

Job Lifecycle

Submitting a Job

flo processing submit pipeline.yaml -n analytics
# Output: Job submitted: job-1

On submission, the job enters the RUNNING state and begins polling its sources immediately.

Checking Status

flo processing status job-1 -n analytics

Returns JSON:

{
  "job_id": "job-1",
  "name": "order-analytics",
  "status": "RUNNING",
  "parallelism": 4,
  "batch_size": 200,
  "records_processed": 15420,
  "created_at_ms": 1700000000000
}

Listing Jobs

flo processing list -n analytics

Stopping a Job

Stops the job gracefully. The job can be restored later:

flo processing stop job-1 -n analytics

Cancelling a Job

Cancels the job permanently:

flo processing cancel job-1 -n analytics

Job States

Savepoints

# Create a savepoint
flo processing savepoint job-1 -n analytics
 
# Restore from a savepoint
flo processing restore job-1 sp-1 -n analytics

Rescaling

Change the parallelism of a running job:

flo processing rescale job-1 8 -n analytics

Namespace Isolation

Jobs are scoped to namespaces, providing complete isolation:

• The same job name in different namespaces are independent jobs
• list returns only jobs in the specified namespace
• Pipeline data flowsonly within the namespace scope
• Source streams and sink targets are resolved within the job's namespace

# These are two separate, independent jobs
flo processing submit job.yaml -n production
flo processing submit job.yaml -n staging

Persistence

Job state survives node restarts:

• Submitted jobs persist and resume on restart — the execution pipeline is rebuilt for RUNNING jobs, not just the registry entry, so a job keeps consuming after restart instead of becoming an idle zombie
• Stream-source pipelines resume from a durable source cursor rather than re-reading from the beginning. The cursor is persisted as a processing_checkpoint entry (debounced to ≥1s per pipeline to bound write amplification)
• Recovery is at-least-once: a crash loses at most the debounce window, which redelivers — it never skips records
• Stopped and cancelled job states are preserved (they stay idle on restart)
• Rescaled parallelism is remembered
• New job IDs don't collide with pre-restart IDs

All mutations (submit, stop, cancel, status updates) are persisted through Raft and replayed on startup.

Processing Metrics

The engine collects per-operator and per-pipeline metrics:

Related Docs

• Streams — source and sink behavior for stream-backed pipelines
• Time-Series — TS query and write semantics
• KV — KV storage used by kv_lookup operator and KV sinks
• Queues — queue sinks for task pipelines
• Actions — worker-hosted action registration and invocation
• Workflows — multi-step orchestration over actions