inputs

An input is where the messages of a pipeline come from. A pipeline can have several inputs. kayak merges them into one stream. Each input has its own task, so a busy input cannot starve a slow input or an input on a timer. When one input fails, kayak reports the failure and the other inputs continue. The pipeline stops only when its last input stops.

Every input kind accepts three more fields. They are on the wrapper of the input, so the tables below do not show them:

• buffer collects messages before the transforms see them: by count, by time, or by the first of the two. It never sends an empty batch. Its window opens at the first message, not at a clock boundary. Thus it limits the latency, and it does not give a fixed cadence. See buffering an input.
• envelope adds what the input knows about a message to the message, as ordinary JSON fields. Without it, kayak passes messages on as they arrive. See message metadata.
• ack sets when the input acknowledges a message to its broker. Only inputs with a broker that can tell the difference accept it. The other inputs refuse to build with it. See acknowledging an input.

max_batch, on the inputs that have it, is different from buffer. It never waits. It takes one message and then adds the messages that already arrived. Thus a quiet topic gives batches of one, whatever the limit is. Only a backlog fills a batch. Increase max_batch to let a consumer catch up on a backlog.

dummy

Sends one generated message at a fixed interval. Use it to test a pipeline without a real source.

Each message contains a value and the current_time at which the input sent it. The payload field sets the type of value. It can be a number from a sine wave or a random sentence.

field	type		description
duration	integer	required	The time between two messages, in s.
amplitude	number	optional	The peak of the sine wave. The value goes from -amplitude to +amplitude. Applies only to the number payload. The default is 1.
payload	number | text	optional	The type of value in each message. number is a number from a sine wave. text is a random sentence. The default is number.
period	number	optional	The time for one full cycle of the sine wave, in s. Applies only to the number payload. The default is 60. The input samples the wave by the clock, so the period does not change with duration.
ack	on_receipt | on_delivery	optional	When the input acknowledges a message to its broker. The default is on_receipt. Only the kafka and mqtt inputs support on_delivery. The mqtt input requires a qos of at_least_once or higher for it. On all other inputs, on_delivery fails to build.
buffer	static | tumbling | batch	optional	Collect messages from this input into batches before the transforms. Use a count (static), a time (tumbling) or the first of the two (batch). The buffer never sends an empty batch. Available on all input types. This is not the buffer transform.
envelope	merge | wrap	optional	Add metadata about the source of each message, for example the subject, the topic or the partition. The "metadata" section lists the fields. Available on all input types. If you do not set it, the input sends each message without changes.

buffer — type: "static"

field	type		description
size	integer	required	The number of messages in a batch.

buffer — type: "tumbling"

field	type		description
window_seconds	integer	required	The time to collect messages, in s, from the first message.

buffer — type: "batch"

field	type		description
size	integer	required	The number of messages that closes the batch immediately.
window_seconds	integer	required	The maximum time to wait, in s, from the first message in the batch.

envelope — type: "merge"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.

envelope — type: "wrap"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.
payload	string	optional	The field for the original payload. The default is value.

metadata — what this input attaches to a message when its envelope is set.

field	holds
pipeline	The id of the pipeline that read the message.
input	The type of the input that read the message, for example nats.
received_at	The time when kayak read the message, as RFC 3339. This is the arrival time at this pipeline. It is not the time of the event in the message.

http

Accepts messages that are posted to the endpoint of the pipeline, POST /api/pipelines/{id}/messages.

kayak makes the endpoint from the pipeline id. The endpoint is available when the pipeline runs. The body is one JSON message or an array of messages. An array becomes one batch. A pipeline can have only one http input. A second http input fails to build.

field	type		description
auth	bearer | header	optional	The credential that a post must have. If you do not set it, the endpoint accepts all posts. A post without the correct credential gets 401.
capacity	integer	optional	The maximum number of posted batches in the queue before the pipeline. The default is 1024. When the queue is full, the endpoint refuses a post with 503.
ack	on_receipt | on_delivery	optional	When the input acknowledges a message to its broker. The default is on_receipt. Only the kafka and mqtt inputs support on_delivery. The mqtt input requires a qos of at_least_once or higher for it. On all other inputs, on_delivery fails to build.
buffer	static | tumbling | batch	optional	Collect messages from this input into batches before the transforms. Use a count (static), a time (tumbling) or the first of the two (batch). The buffer never sends an empty batch. Available on all input types. This is not the buffer transform.
envelope	merge | wrap	optional	Add metadata about the source of each message, for example the subject, the topic or the partition. The "metadata" section lists the fields. Available on all input types. If you do not set it, the input sends each message without changes.

auth — type: "bearer"

field	type		description
token	string	required	The token. Use a ${NAME} reference, so that the config file keeps only the name and the secret store keeps the value.

auth — type: "header"

field	type		description
name	string	required	The name of the header. The http input compares the name without case. On an http input, the name must not be a header that an envelope copies into the messages.
value	string	required	The exact value of the header. Use a ${NAME} reference.

buffer — type: "static"

field	type		description
size	integer	required	The number of messages in a batch.

buffer — type: "tumbling"

field	type		description
window_seconds	integer	required	The time to collect messages, in s, from the first message.

buffer — type: "batch"

field	type		description
size	integer	required	The number of messages that closes the batch immediately.
window_seconds	integer	required	The maximum time to wait, in s, from the first message in the batch.

envelope — type: "merge"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.

envelope — type: "wrap"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.
payload	string	optional	The field for the original payload. The default is value.

metadata — what this input attaches to a message when its envelope is set.

field	holds
pipeline	The id of the pipeline that read the message.
input	The type of the input that read the message, for example nats.
received_at	The time when kayak read the message, as RFC 3339. This is the arrival time at this pipeline. It is not the time of the event in the message.
method	The HTTP method of the request.
remote_addr	The address that sent the request, when the server knows it.
headers	The request headers from a fixed list: content-type, user-agent, x-request-id, x-correlation-id and traceparent. The input drops all other headers. Thus, a credential header such as authorization or x-api-key never goes into the message.

kafka

Consumes JSON messages from a kafka topic.

The input skips a payload that is not JSON and writes a warning to the log. The consumer connects on the first read and joins a consumer group. Kafka keeps the read position of the group between restarts.

field	type		description
connection	kafka connection	required	The name of the kafka connection to consume from. Declare the connection in the connections file.
topic	string	required	The topic to consume from.
group	string	required	The consumer group id. Kafka keeps one read position for each group. Two pipelines in the same group divide the topic between them. Two pipelines in different groups each get all messages.
max_batch	integer	optional	The maximum number of messages in one batch. The default is 1. The input puts only records that are already received into a batch. It does not wait for more records, so a high value does not add latency on a quiet topic. Increase it when the consumer reads a backlog. Each batch has a fixed cost in the pipeline and in each downstream pipeline.
start_at	earliest | latest	optional	The start position when the group has no committed position. earliest reads the topic from the start. latest reads only new messages. The default is latest.
ack	on_receipt | on_delivery	optional	When the input acknowledges a message to its broker. The default is on_receipt. Only the kafka and mqtt inputs support on_delivery. The mqtt input requires a qos of at_least_once or higher for it. On all other inputs, on_delivery fails to build.
buffer	static | tumbling | batch	optional	Collect messages from this input into batches before the transforms. Use a count (static), a time (tumbling) or the first of the two (batch). The buffer never sends an empty batch. Available on all input types. This is not the buffer transform.
envelope	merge | wrap	optional	Add metadata about the source of each message, for example the subject, the topic or the partition. The "metadata" section lists the fields. Available on all input types. If you do not set it, the input sends each message without changes.

buffer — type: "static"

field	type		description
size	integer	required	The number of messages in a batch.

buffer — type: "tumbling"

field	type		description
window_seconds	integer	required	The time to collect messages, in s, from the first message.

buffer — type: "batch"

field	type		description
size	integer	required	The number of messages that closes the batch immediately.
window_seconds	integer	required	The maximum time to wait, in s, from the first message in the batch.

envelope — type: "merge"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.

envelope — type: "wrap"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.
payload	string	optional	The field for the original payload. The default is value.

metadata — what this input attaches to a message when its envelope is set.

field	holds
pipeline	The id of the pipeline that read the message.
input	The type of the input that read the message, for example nats.
received_at	The time when kayak read the message, as RFC 3339. This is the arrival time at this pipeline. It is not the time of the event in the message.
connection	The name of the connection that the input consumed from.
topic	The topic of the record.
partition	The partition of the record in the topic.
offset	The offset of the record in the partition. With topic and partition, it identifies the record.
key	The key of the record as a string, or null when the record has no key.
timestamp	The timestamp of the record, as RFC 3339, when kafka gives one.

nats

Subscribes to a nats subject and parses each message as JSON.

The input skips a payload that is not JSON and writes a warning to the log. The input opens the connection on the first read.

field	type		description
connection	nats connection	required	The name of the nats connection to subscribe on. Declare the connection in the connections file.
subject	string	required	The subject to subscribe to.
max_batch	integer	optional	The maximum number of messages in one batch. The default is 1. The input puts only messages that are already received into a batch. It does not wait for more messages, so a high value does not add latency on a quiet subject.
ack	on_receipt | on_delivery	optional	When the input acknowledges a message to its broker. The default is on_receipt. Only the kafka and mqtt inputs support on_delivery. The mqtt input requires a qos of at_least_once or higher for it. On all other inputs, on_delivery fails to build.
buffer	static | tumbling | batch	optional	Collect messages from this input into batches before the transforms. Use a count (static), a time (tumbling) or the first of the two (batch). The buffer never sends an empty batch. Available on all input types. This is not the buffer transform.
envelope	merge | wrap	optional	Add metadata about the source of each message, for example the subject, the topic or the partition. The "metadata" section lists the fields. Available on all input types. If you do not set it, the input sends each message without changes.

buffer — type: "static"

field	type		description
size	integer	required	The number of messages in a batch.

buffer — type: "tumbling"

field	type		description
window_seconds	integer	required	The time to collect messages, in s, from the first message.

buffer — type: "batch"

field	type		description
size	integer	required	The number of messages that closes the batch immediately.
window_seconds	integer	required	The maximum time to wait, in s, from the first message in the batch.

envelope — type: "merge"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.

envelope — type: "wrap"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.
payload	string	optional	The field for the original payload. The default is value.

metadata — what this input attaches to a message when its envelope is set.

field	holds
pipeline	The id of the pipeline that read the message.
input	The type of the input that read the message, for example nats.
received_at	The time when kayak read the message, as RFC 3339. This is the arrival time at this pipeline. It is not the time of the event in the message.
connection	The name of the connection that received the message.
subject	The subject that the message was published to. This is the full subject, not the subscription pattern. For example, when you subscribe to *.temperature, this field holds the name of the machine.
reply	The reply subject, when the publisher set one. Otherwise null.
headers	The nats headers, as an object of arrays.

pipeline

Reads the output of another pipeline. Use it to connect pipelines into a graph.

Many pipelines can read from the same upstream. Each of them gets all batches. The upstream must exist when kayak creates this pipeline, so declare the upstream first in the config file.

field	type		description
upstream	pipeline id	required	The id of the pipeline to read from.
ack	on_receipt | on_delivery	optional	When the input acknowledges a message to its broker. The default is on_receipt. Only the kafka and mqtt inputs support on_delivery. The mqtt input requires a qos of at_least_once or higher for it. On all other inputs, on_delivery fails to build.
buffer	static | tumbling | batch	optional	Collect messages from this input into batches before the transforms. Use a count (static), a time (tumbling) or the first of the two (batch). The buffer never sends an empty batch. Available on all input types. This is not the buffer transform.
envelope	merge | wrap	optional	Add metadata about the source of each message, for example the subject, the topic or the partition. The "metadata" section lists the fields. Available on all input types. If you do not set it, the input sends each message without changes.

buffer — type: "static"

field	type		description
size	integer	required	The number of messages in a batch.

buffer — type: "tumbling"

field	type		description
window_seconds	integer	required	The time to collect messages, in s, from the first message.

buffer — type: "batch"

field	type		description
size	integer	required	The number of messages that closes the batch immediately.
window_seconds	integer	required	The maximum time to wait, in s, from the first message in the batch.

envelope — type: "merge"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.

envelope — type: "wrap"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.
payload	string	optional	The field for the original payload. The default is value.

metadata — what this input attaches to a message when its envelope is set.

field	holds
pipeline	The id of the pipeline that read the message.
input	The type of the input that read the message, for example nats.
received_at	The time when kayak read the message, as RFC 3339. This is the arrival time at this pipeline. It is not the time of the event in the message.
upstream	The id of the pipeline that sent the batch. Metadata from an upstream input stays in the message. This input does not replace it.

mqtt

Subscribes to an mqtt topic and parses each message as JSON.

The topic can be a filter with the mqtt wildcards + and #. The input skips a payload that is not JSON and writes a warning to the log.

The input opens the connection on the first read. kayak makes the client id from the pipeline id and the topic. You cannot set the client id.

field	type		description
connection	mqtt connection	required	The name of the mqtt connection to subscribe on. Declare the connection in the connections file.
topic	string	required	The topic or topic filter to subscribe to.
max_batch	integer	optional	The maximum number of messages in one batch. The default is 1. The input puts only messages that are already received into a batch. It does not wait for more messages, so a high value does not add latency on a quiet topic.
qos	at_most_once | at_least_once | exactly_once	optional	The quality of service for the subscription. The default is at_most_once. ack: on_delivery requires at_least_once or exactly_once.
ack	on_receipt | on_delivery	optional	When the input acknowledges a message to its broker. The default is on_receipt. Only the kafka and mqtt inputs support on_delivery. The mqtt input requires a qos of at_least_once or higher for it. On all other inputs, on_delivery fails to build.
buffer	static | tumbling | batch	optional	Collect messages from this input into batches before the transforms. Use a count (static), a time (tumbling) or the first of the two (batch). The buffer never sends an empty batch. Available on all input types. This is not the buffer transform.
envelope	merge | wrap	optional	Add metadata about the source of each message, for example the subject, the topic or the partition. The "metadata" section lists the fields. Available on all input types. If you do not set it, the input sends each message without changes.

buffer — type: "static"

field	type		description
size	integer	required	The number of messages in a batch.

buffer — type: "tumbling"

field	type		description
window_seconds	integer	required	The time to collect messages, in s, from the first message.

buffer — type: "batch"

field	type		description
size	integer	required	The number of messages that closes the batch immediately.
window_seconds	integer	required	The maximum time to wait, in s, from the first message in the batch.

envelope — type: "merge"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.

envelope — type: "wrap"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.
payload	string	optional	The field for the original payload. The default is value.

metadata — what this input attaches to a message when its envelope is set.

field	holds
pipeline	The id of the pipeline that read the message.
input	The type of the input that read the message, for example nats.
received_at	The time when kayak read the message, as RFC 3339. This is the arrival time at this pipeline. It is not the time of the event in the message.
connection	The name of the connection that received the message.
topic	The full topic of the message. Use it when the input subscribes to a filter with + or # wildcards.
qos	The quality of service of the delivery.
retain	True when the broker sent the retained message of the topic. False for a live publish.

redis

Subscribes to a redis channel and parses each message as JSON.

The input skips a payload that is not JSON and writes a warning to the log. The input opens the connection on the first read.

The input uses SUBSCRIBE, so the channel name must be exact. Patterns are not supported. Redis pub/sub does not send a message again. When the input is not connected, it does not receive the messages that are published.

field	type		description
connection	redis connection	required	The name of the redis connection to subscribe on. Declare the connection in the connections file.
channel	string	required	The channel to subscribe to.
max_batch	integer	optional	The maximum number of messages in one batch. The default is 1. The input puts only messages that are already received into a batch. It does not wait for more messages, so a high value does not add latency on a quiet channel.
ack	on_receipt | on_delivery	optional	When the input acknowledges a message to its broker. The default is on_receipt. Only the kafka and mqtt inputs support on_delivery. The mqtt input requires a qos of at_least_once or higher for it. On all other inputs, on_delivery fails to build.
buffer	static | tumbling | batch	optional	Collect messages from this input into batches before the transforms. Use a count (static), a time (tumbling) or the first of the two (batch). The buffer never sends an empty batch. Available on all input types. This is not the buffer transform.
envelope	merge | wrap	optional	Add metadata about the source of each message, for example the subject, the topic or the partition. The "metadata" section lists the fields. Available on all input types. If you do not set it, the input sends each message without changes.

buffer — type: "static"

field	type		description
size	integer	required	The number of messages in a batch.

buffer — type: "tumbling"

field	type		description
window_seconds	integer	required	The time to collect messages, in s, from the first message.

buffer — type: "batch"

field	type		description
size	integer	required	The number of messages that closes the batch immediately.
window_seconds	integer	required	The maximum time to wait, in s, from the first message in the batch.

envelope — type: "merge"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.

envelope — type: "wrap"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.
payload	string	optional	The field for the original payload. The default is value.

metadata — what this input attaches to a message when its envelope is set.

field	holds
pipeline	The id of the pipeline that read the message.
input	The type of the input that read the message, for example nats.
received_at	The time when kayak read the message, as RFC 3339. This is the arrival time at this pipeline. It is not the time of the event in the message.
connection	The name of the connection that received the message.
channel	The channel that the message was published to.

opcua

Subscribes to variables on an OPC UA server and sends one message for each change of a value.

The input makes a subscription with one monitored item for each node. The server sends a value when it changes. The input does not poll. A tag that does not change sends no messages. publish_interval_ms sets how frequently the server can send. It does not set how frequently the server samples.

Each message is one reading. It contains the tag and the value:

json { "node": "ns=2;s=Machine1.Temperature", "name": "temperature", "value": 21.5, "status": "Good", "source_timestamp": "2026-01-01T12:00:00.123Z", "server_timestamp": "2026-01-01T12:00:00.130Z" }

status is the quality of the reading and is always present. A failed sensor sends a Bad... status with a null value. Use a filter to remove these readings. source_timestamp is the time at which the device produced the value. Use it to reduce or partition. The received_at field of the envelope is the time at which kayak read the value.

Set nodes, browse or both. One of them is required. The input subscribes to a node only one time, also when two settings name it.

field	type		description
connection	opcua connection	required	The name of the opcua connection to subscribe on. Declare the connection in the connections file.
browse	object	optional	A node to browse. The input subscribes to each variable under it.
deadband	number	optional	The smallest change of a value that the server reports, in the units of the value. If you do not set it, the server reports each change. The server applies the deadband, so it decreases network traffic. It applies only to numeric nodes. The server reports each change of a string or a boolean.
max_batch	integer	optional	The maximum number of messages in one batch. The default is 1. One publish from the server contains each node that changed in the interval. With 200 tags at 1 Hz and the default, the pipeline handles 200 batches each second. A higher value decreases this cost. The input puts only changes that are already received into a batch.
nodes	list of opcua node	optional	The nodes to subscribe to, one entry for each node.
publish_interval_ms	integer	optional	The interval at which the server can send a group of changes, in ms. The default is 1000. This value sets the longest time that a change waits.
queue_size	integer	optional	The number of samples that the server keeps for one node between two publishes. The default is 1. With 1, the server sends only the latest value of a node that changes two times in one interval. Increase it, together with sampling_interval_ms, when you need each sample.
sampling_interval_ms	integer	optional	The interval at which the server samples each node, in ms. If you do not set it, the server samples at the publish interval. Set a smaller value to get more readings between two publishes.
ack	on_receipt | on_delivery	optional	When the input acknowledges a message to its broker. The default is on_receipt. Only the kafka and mqtt inputs support on_delivery. The mqtt input requires a qos of at_least_once or higher for it. On all other inputs, on_delivery fails to build.
buffer	static | tumbling | batch	optional	Collect messages from this input into batches before the transforms. Use a count (static), a time (tumbling) or the first of the two (batch). The buffer never sends an empty batch. Available on all input types. This is not the buffer transform.
envelope	merge | wrap	optional	Add metadata about the source of each message, for example the subject, the topic or the partition. The "metadata" section lists the fields. Available on all input types. If you do not set it, the input sends each message without changes.

browse

field	type		description
root	string	required	The id of the node to browse under, in the same notation as node_id. This is usually a folder, for example ns=2;s=Machine1. The input subscribes to each variable under it. It follows folders and objects but does not subscribe to them.
depth	integer	optional	The number of levels below the root to follow. The default is 3. The value 0 is not permitted. There is no value for "all levels", because the address space of a plant server can have thousands of nodes.

nodes — each entry

field	type		description
node_id	string	required	The id of the node, in OPC UA notation. Use ns=2;s=Machine1.Temperature for a string identifier and ns=2;i=1042 for a numeric identifier. Use g= for a GUID and b= for an opaque identifier. A node id with no ns= is in namespace 0, the namespace of the server.
name	string	optional	The name of the node in the messages. The default is the node id. Set a readable name to use in a group_by or a column mapping.

buffer — type: "static"

field	type		description
size	integer	required	The number of messages in a batch.

buffer — type: "tumbling"

field	type		description
window_seconds	integer	required	The time to collect messages, in s, from the first message.

buffer — type: "batch"

field	type		description
size	integer	required	The number of messages that closes the batch immediately.
window_seconds	integer	required	The maximum time to wait, in s, from the first message in the batch.

envelope — type: "merge"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.

envelope — type: "wrap"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.
payload	string	optional	The field for the original payload. The default is value.

metadata — what this input attaches to a message when its envelope is set.

field	holds
pipeline	The id of the pipeline that read the message.
input	The type of the input that read the message, for example nats.
received_at	The time when kayak read the message, as RFC 3339. This is the arrival time at this pipeline. It is not the time of the event in the message.
connection	The name of the connection of the session. The node of the reading is not metadata. It is always in the message, as node and name.

indu

Reads live sensors and streams from Indu Cloud through /api/v1/live/sse, with the API key of the connection.

Name sensors and streams with the ids that the platform uses. Do not use UUIDs. The input finds the names through /api/v1 on the first read. If the key cannot find or see a name, the input reports an error. It then tries again after a pause.

Each reading is one message, for example {"kind": "sensor", "name": "press-3/temperature", "value": 71.2, "at": …}. The message also contains the ids of the platform. When the connection drops, the input connects again with backoff. If the input cannot read all readings, it reports an error.

field	type		description
connection	indu connection	required	The name of the indu connection to read through. Declare the connection in the connections file.
backfill	boolean	optional	Send the latest value of each series before the live readings. The default is true. Thus a restarted pipeline has a value for each series immediately.
max_batch	integer	optional	The maximum number of readings in one batch. The default is 1. The input puts only readings that are already received into a batch. It does not wait for more readings.
sensors	list of string	optional	The sensors to read, as <device>/<sensor>, for example press-3/temperature. Use the ids that the platform uses. kayak divides the name at the first /.
streams	list of string	optional	The streams to read, by the name they were written under, for example press-3/oee. For a stream that the platform calculates, use its display name.
ack	on_receipt | on_delivery	optional	When the input acknowledges a message to its broker. The default is on_receipt. Only the kafka and mqtt inputs support on_delivery. The mqtt input requires a qos of at_least_once or higher for it. On all other inputs, on_delivery fails to build.
buffer	static | tumbling | batch	optional	Collect messages from this input into batches before the transforms. Use a count (static), a time (tumbling) or the first of the two (batch). The buffer never sends an empty batch. Available on all input types. This is not the buffer transform.
envelope	merge | wrap	optional	Add metadata about the source of each message, for example the subject, the topic or the partition. The "metadata" section lists the fields. Available on all input types. If you do not set it, the input sends each message without changes.

buffer — type: "static"

field	type		description
size	integer	required	The number of messages in a batch.

buffer — type: "tumbling"

field	type		description
window_seconds	integer	required	The time to collect messages, in s, from the first message.

buffer — type: "batch"

field	type		description
size	integer	required	The number of messages that closes the batch immediately.
window_seconds	integer	required	The maximum time to wait, in s, from the first message in the batch.

envelope — type: "merge"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.

envelope — type: "wrap"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.
payload	string	optional	The field for the original payload. The default is value.

metadata — what this input attaches to a message when its envelope is set.

field	holds
pipeline	The id of the pipeline that read the message.
input	The type of the input that read the message, for example nats.
received_at	The time when kayak read the message, as RFC 3339. This is the arrival time at this pipeline. It is not the time of the event in the message.
connection	The name of the connection that the input read from.
event	The platform event of the message: reading for a sensor, stream_reading for a stream.

postgres

Reads a postgres table, view or query at an interval and sends each row as a message.

The input runs a query each interval_secs. In snapshot mode, it reads the full relation. In incremental mode, it reads only the rows after the last read, by a column that increases. Each row becomes one JSON object with the column names as fields. Postgres makes the object with row_to_json. A timestamp is ISO 8601, a numeric keeps its digits, and a jsonb column is a nested value:

json {"id": 42, "sensor": "press-3", "value": 21.5, "recorded_at": "2026-01-01T12:00:00.123456+00:00"}

An incremental read is at-least-once across a restart. The input keeps the watermark in memory, and after a restart it starts again from start_from. An incremental read does not see a deleted row. Put an index on the column that it follows, or each read scans the full table.

field	type		description
connection	postgres connection	required	The name of the postgres connection to read through. Declare the connection in the connections file.
interval_secs	integer	required	The time between two reads, in seconds. The time starts at the end of a read. Thus, two reads never overlap. The first read occurs when the pipeline starts.
mode	snapshot | incremental	required	Whether each read returns all rows (snapshot) or only the rows after the last read (incremental).
columns	list of string	optional	The columns to read, in order. Leave it empty to read all columns. For an incremental input, the list must include field.
max_batch	integer	optional	The maximum number of rows in one batch. The default is 1. The input puts rows that it already read into a batch. It does not wait for more rows to fill a batch.
page_size	integer	optional	The maximum number of rows that one query returns. The default is 1000. When a page is full, the input reads the next page immediately. The interval starts after a page that is not full. A snapshot ignores this field and reads all rows in one query.
query	string	optional	A SELECT to read in place of a table. The input puts the query in a subquery. It adds the cursor condition, the ORDER BY and the page limit outside the subquery. Thus, the query needs no placeholder and no ORDER BY. Write one statement with no semicolon at the end. You can use all SQL that the server accepts in a subquery, for example WITH.
table	string	optional	The table or view to read, as name or schema.name. Give exactly one of table and query.
ack	on_receipt | on_delivery	optional	When the input acknowledges a message to its broker. The default is on_receipt. Only the kafka and mqtt inputs support on_delivery. The mqtt input requires a qos of at_least_once or higher for it. On all other inputs, on_delivery fails to build.
buffer	static | tumbling | batch	optional	Collect messages from this input into batches before the transforms. Use a count (static), a time (tumbling) or the first of the two (batch). The buffer never sends an empty batch. Available on all input types. This is not the buffer transform.
envelope	merge | wrap	optional	Add metadata about the source of each message, for example the subject, the topic or the partition. The "metadata" section lists the fields. Available on all input types. If you do not set it, the input sends each message without changes.

mode — type: "snapshot"

This component takes no configuration.

mode — type: "incremental"

field	type		description
field	string	required	The column that the input follows. The watermark is the highest value that the input sent. Each read asks for the rows above the watermark. The input does not read rows where the column is null.
lag_secs	integer	optional	For a timestamp column: the time to stay behind the current time, in seconds. A later read gets the rows that are less than this time before now(). This gives late transactions time to commit. Do not use it with a numeric column. The server refuses the query.
start_from	oldest | newest	optional	Where the first read starts. With newest, the input reads only the rows added after the pipeline started. With oldest, it reads all rows first and then follows the table. The default is newest.

buffer — type: "static"

field	type		description
size	integer	required	The number of messages in a batch.

buffer — type: "tumbling"

field	type		description
window_seconds	integer	required	The time to collect messages, in s, from the first message.

buffer — type: "batch"

field	type		description
size	integer	required	The number of messages that closes the batch immediately.
window_seconds	integer	required	The maximum time to wait, in s, from the first message in the batch.

envelope — type: "merge"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.

envelope — type: "wrap"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.
payload	string	optional	The field for the original payload. The default is value.

metadata — what this input attaches to a message when its envelope is set.

field	holds
pipeline	The id of the pipeline that read the message.
input	The type of the input that read the message, for example nats.
received_at	The time when kayak read the message, as RFC 3339. This is the arrival time at this pipeline. It is not the time of the event in the message.
connection	The name of the connection that the input read from.
polled_at	The start time of the read that returned the row, as RFC 3339. All rows of one read have the same value. Use it to tell one snapshot from the next. The time comes from the clock of kayak, not from the database server.

clickhouse

Reads a ClickHouse table, view or query at an interval and sends each row as a message. It uses the HTTP interface of ClickHouse.

ClickHouse sends the rows as JSONEachRow. A DateTime is ISO 8601, an Int64 is a number, and a Decimal keeps its digits. The modes, the watermark and the limits of an incremental read are the same as on the postgres input.

field	type		description
connection	clickhouse connection	required	The name of the clickhouse connection to read through. Declare the connection in the connections file.
interval_secs	integer	required	The time between two reads, in seconds. The time starts at the end of a read. Thus, two reads never overlap. The first read occurs when the pipeline starts.
mode	snapshot | incremental	required	Whether each read returns all rows (snapshot) or only the rows after the last read (incremental).
columns	list of string	optional	The columns to read, in order. Leave it empty to read all columns. For an incremental input, the list must include field.
max_batch	integer	optional	The maximum number of rows in one batch. The default is 1. The input puts rows that it already read into a batch. It does not wait for more rows to fill a batch.
page_size	integer	optional	The maximum number of rows that one query returns. The default is 1000. When a page is full, the input reads the next page immediately. The interval starts after a page that is not full. A snapshot ignores this field and reads all rows in one query.
query	string	optional	A SELECT to read in place of a table. The input puts the query in a subquery. It adds the cursor condition, the ORDER BY and the page limit outside the subquery. Thus, the query needs no placeholder and no ORDER BY. Write one statement with no semicolon at the end. You can use all SQL that the server accepts in a subquery, for example WITH.
table	string	optional	The table or view to read, as name or schema.name. Give exactly one of table and query.
ack	on_receipt | on_delivery	optional	When the input acknowledges a message to its broker. The default is on_receipt. Only the kafka and mqtt inputs support on_delivery. The mqtt input requires a qos of at_least_once or higher for it. On all other inputs, on_delivery fails to build.
buffer	static | tumbling | batch	optional	Collect messages from this input into batches before the transforms. Use a count (static), a time (tumbling) or the first of the two (batch). The buffer never sends an empty batch. Available on all input types. This is not the buffer transform.
envelope	merge | wrap	optional	Add metadata about the source of each message, for example the subject, the topic or the partition. The "metadata" section lists the fields. Available on all input types. If you do not set it, the input sends each message without changes.

mode — type: "snapshot"

This component takes no configuration.

mode — type: "incremental"

field	type		description
field	string	required	The column that the input follows. The watermark is the highest value that the input sent. Each read asks for the rows above the watermark. The input does not read rows where the column is null.
lag_secs	integer	optional	For a timestamp column: the time to stay behind the current time, in seconds. A later read gets the rows that are less than this time before now(). This gives late transactions time to commit. Do not use it with a numeric column. The server refuses the query.
start_from	oldest | newest	optional	Where the first read starts. With newest, the input reads only the rows added after the pipeline started. With oldest, it reads all rows first and then follows the table. The default is newest.

buffer — type: "static"

field	type		description
size	integer	required	The number of messages in a batch.

buffer — type: "tumbling"

field	type		description
window_seconds	integer	required	The time to collect messages, in s, from the first message.

buffer — type: "batch"

field	type		description
size	integer	required	The number of messages that closes the batch immediately.
window_seconds	integer	required	The maximum time to wait, in s, from the first message in the batch.

envelope — type: "merge"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.

envelope — type: "wrap"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.
payload	string	optional	The field for the original payload. The default is value.

metadata — what this input attaches to a message when its envelope is set.

field	holds
pipeline	The id of the pipeline that read the message.
input	The type of the input that read the message, for example nats.
received_at	The time when kayak read the message, as RFC 3339. This is the arrival time at this pipeline. It is not the time of the event in the message.
connection	The name of the connection that the input read from.
polled_at	The start time of the read that returned the row, as RFC 3339. All rows of one read have the same value. Use it to tell one snapshot from the next. The time comes from the clock of kayak, not from the database server.

http_poll

Gets a url at an interval and sends the full reply each time.

Each read is a GET. A reply that is an array gives one message for each element. Any other reply gives one message. Use items for a reply that has the records inside it, for example {"data": {"machines": [...]}}.

Use this input for reference data, for example a list of machines or recipes that changes rarely. The input has no watermark and no pages. Each read sends all records again. Send them to an output that writes the latest value for each key.

A read fails when the url cannot be reached, when the status is not 2xx, when the body is not JSON, or when items finds nothing. The input reports the failure one time and tries again with backoff. The interval starts again after the next successful read.

field	type		description
url	string	required	The url to get, for example https://erp.example.com/api/machines.
interval_secs	integer	required	The time between two reads, in s. The time starts at the end of one read. The first read occurs when the pipeline starts.
auth	bearer | header	optional	The credential that the input sends. If you do not set it, the input sends no credential.
items	string	optional	The position of the records in the reply, as a JSON pointer. For example, /data/machines reads the array at data.machines. If you do not set it, the input uses the full reply. An array gives one message for each element. Any other value gives one message.
max_batch	integer	optional	The maximum number of messages in one batch. The default is 1. The input puts only messages that are already read into a batch. It does not wait for a batch to fill.
timeout_seconds	integer	optional	The maximum time for one request, in s. The default is 30.
ack	on_receipt | on_delivery	optional	When the input acknowledges a message to its broker. The default is on_receipt. Only the kafka and mqtt inputs support on_delivery. The mqtt input requires a qos of at_least_once or higher for it. On all other inputs, on_delivery fails to build.
buffer	static | tumbling | batch	optional	Collect messages from this input into batches before the transforms. Use a count (static), a time (tumbling) or the first of the two (batch). The buffer never sends an empty batch. Available on all input types. This is not the buffer transform.
envelope	merge | wrap	optional	Add metadata about the source of each message, for example the subject, the topic or the partition. The "metadata" section lists the fields. Available on all input types. If you do not set it, the input sends each message without changes.

auth — type: "bearer"

field	type		description
token	string	required	The token. Use a ${NAME} reference, so that the config file keeps only the name and the secret store keeps the value.

auth — type: "header"

field	type		description
name	string	required	The name of the header. The http input compares the name without case. On an http input, the name must not be a header that an envelope copies into the messages.
value	string	required	The exact value of the header. Use a ${NAME} reference.

buffer — type: "static"

field	type		description
size	integer	required	The number of messages in a batch.

buffer — type: "tumbling"

field	type		description
window_seconds	integer	required	The time to collect messages, in s, from the first message.

buffer — type: "batch"

field	type		description
size	integer	required	The number of messages that closes the batch immediately.
window_seconds	integer	required	The maximum time to wait, in s, from the first message in the batch.

envelope — type: "merge"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.

envelope — type: "wrap"

field	type		description
meta	string	optional	The field for the metadata object. The default is _meta.
payload	string	optional	The field for the original payload. The default is value.

metadata — what this input attaches to a message when its envelope is set.

field	holds
pipeline	The id of the pipeline that read the message.
input	The type of the input that read the message, for example nats.
received_at	The time when kayak read the message, as RFC 3339. This is the arrival time at this pipeline. It is not the time of the event in the message.
url	The url that the input read the message from, with no username and no password.
polled_at	The start time of the read that returned the message, as RFC 3339. All messages of one read have the same value. Use it to tell one snapshot from the next.