kayak is a stream processor written in Rust. You write pipelines in a config file and keep the file in version control. You run the container image with that file. That is the complete deployment.
$ docker run -p 6767:6767 -v "$PWD:/kayak" \
ghcr.io/niclasgrahm/kayak --config /kayak/config.yamlone rust binary · JSON or YAML · no cluster · no database of its own
- id: warm_sensors inputs: - type: nats connection: plant-nats subject: sensors.> max_batch: 500 transforms: - type: filter conditions: - { type: numeric, field: value, operator: greater_than, value: 30 } outputs: - type: clickhouse connection: analytics table: warm_readings
inputs → transforms → outputs. The names plant-nats and analytics are connections. You declare them one time, in config.connections.yaml beside this file.A pipeline is inputs → transforms → outputs, and all three are arrays. kayak merges the inputs into one stream. The transforms run in the order of the config. Each output gets every batch.
The card on the right shows one pipeline. The web UI draws it from the same config: the config in three tabs, a throughput chart, and the batches that went through.
An input reads a subject, a topic, a channel, a table or a set of OPC UA nodes. The http input receives posts. Each input gives the pipeline batches of plain JSON. You do not declare a schema.
Every input can buffer by count, by time window, or both. Every input can add an envelope with its metadata, for example the subject, the partition or the offset. The envelope is ordinary JSON fields.
natskafkamqttredisopcuahttphttp_pollpostgresclickhouseindupipelinedummyA transform is a small step that does one thing. filter keeps or drops messages. map copies, casts and calculates fields. reducer aggregates with group_by. The streaming transforms keep state per key. Use script when configuration is not sufficient.
Every transform addresses fields by path, for example sensor.id or _meta.subject. Each transform has a policy for a missing field. kayak does not guess.
filtermapreducersplitterbufferrememberrecallscripthttpdeadbandthrottlepivotderiverollingsmoothdetectresamplefeaturesEach output gets every batch. To archive to postgres and also send to kafka, use one pipeline with two outputs. The database outputs map fields to typed columns. The file and s3 outputs rotate parts by rows or by time.
kayak retries an output that it cannot reach at startup. If postgres starts 30 seconds after kayak, the pipeline waits and then continues.
postgresclickhouses3filekafkanatsmqttredishttpindutidepoolstdoutThe pipeline input reads the output of another pipeline. One pipeline reads the broker one time. Many pipelines downstream use its messages: a rollup, an archive, an alert. Your config is a graph of small pipelines.
Each pipeline has its own transforms and outputs. kayak shares each message between them and does not copy it. In the web UI, a line lights up when a batch goes along it.
kayak has no garbage collector. The just bench harness measures the cost of the run loop. These numbers come from an Apple M1 Max, in process, with no network and no disk. They exclude I/O, so they are not end-to-end throughput.
one pipeline, no transforms
one pipeline, batches of 100, one filter
1000 pipelines at the same time, batches of 100, 14 MiB resident
resident memory of one pipeline
Arc).max_batch on a broker input, or an input buffer. kayak then does the work per batch one time for many messages.clickhouse output writes one insert per batch.Each transform does one thing. You connect transforms in a chain, and you connect pipelines in a graph. A message is plain JSON from the input to the output. There is no schema to declare.
pipeline input reads the output of another pipeline. You can make fan-out, fan-in and chains of any depth.remember writes a value to a named bucket. recall reads it, also in a different pipeline.script (rhai) when the other transforms are not sufficient.- id: sensors_10s_avg inputs: - type: pipeline upstream: sensors buffer: { type: tumbling, window_seconds: 10 } transforms: - type: reducer group_by: [sensor, _meta.subject] aggregations: - { function: avg, field: value, as: mean } - { function: max, field: value, as: highest } - { function: count, as: readings } outputs: - type: clickhouse connection: analytics table: sensor_rollups order_by: [sensor] columns: - { name: sensor, type: text } - { name: mean, type: float } - { name: highest, type: float } - { name: readings, type: bigint }
(sensor, subject) pair every 10 s. kayak checks the column mapping when it builds the pipeline.plant-nats: type: nats urls: nats://nats.internal:4222 analytics: type: clickhouse url: https://ch.internal:8443 database: plant user: kayak password: ${CLICKHOUSE_PASSWORD} archive: type: s3 bucket: events region: eu-north-1 access_key_id: ${S3_KEY_ID} secret_access_key: ${S3_SECRET}
${NAME} from the environment or from a secrets file.The reference documents each component and each field. kayak generates the reference from the config types, so it agrees with the server that you run.
natskafkamqttredisopcuahttphttp_pollpostgresclickhouseindupipelinedummyfiltermapreducersplitterbufferrememberrecallhttpscriptdeadbandthrottlepivotderiverollingsmoothdetectresamplefeaturespostgresclickhouses3filekafkanatsmqttredishttpindutidepoolstdoutkafkanatsmqttredispostgresclickhouses3fileopcuaindutidepoolEach input can buffer by count, by time window, or by the first of the two.
With ack: on_delivery, a kafka or mqtt input acknowledges a message after the outputs return.
The postgres and clickhouse outputs map fields to typed columns and create the table.
The file and s3 outputs rotate parts by row count or by time.
Rolling windows, smoothing, rates, resampling and anomaly detection, per key.
The http transform sends messages to a service. It can merge the reply into each message.
The opcua input subscribes to nodes and sends one message for each value change.
The http input gives a pipeline its own endpoint, with an optional credential.
The container image contains the binary and nothing else. You mount your config and name it on the command line. The arguments of the container are the flags of the server.
$ docker run -p 6767:6767 -v "$PWD:/kayak" \
-e CLICKHOUSE_PASSWORD \
ghcr.io/niclasgrahm/kayak \
--config /kayak/config.yaml \
--server-config /kayak/server.yamlconfig.connections.yaml beside the config, so mount the directory. The image is for linux/amd64 and linux/arm64 and runs as uid 10001. In Kubernetes, put the config in a ConfigMap and the secrets in environment variables.auth: type: basic users: ops: { password: ${OPS_PASSWORD}, role: admin } viewer: { password: ${VIEWER_PASSWORD}, role: read } history: retention_secs: 86400 # 0 turns history off
admin can change the graph. read can only look. Accounts come from this file (basic) or from an identity provider (jwt).<config>.connections.<ext> beside the config.${NAME}. kayak reads the environment first. kayak does not start if a name has no value.file outputs. Without it, file outputs do not build.127.0.0.1:6767. The image binds 0.0.0.0:6767.# send messages to a pipeline with an http input $ curl -X POST localhost:6767/api/pipelines/ingest/messages \ -d '[{"sensor":"line1/temp","value":21.5}]' HTTP/1.1 202 Accepted # counters and failure records of the last day $ curl localhost:6767/api/pipelines/ingest/history?resolution=coarse # the complete api, as OpenAPI 3.1 $ curl localhost:6767/api/openapi.json
/api/docs gives the component reference as JSON.Read these limits before you choose kayak for a job.
kayak does not cluster. It has no distributed state and no exactly-once delivery across machines.
State buckets and history are in memory. A restart clears them.
The postgres and clickhouse outputs create a table if it does not exist. They do not change an existing table.
The config format can change between minor versions. The roadmap is in the repository.
The server also has a web UI. Use it to look at the running graph and the messages in each pipeline. You do not need it to run kayak. Read about the web UI.