outputs

Every output on a pipeline gets every batch. An output is initialised when the pipeline is built — which is what makes a wrong password fail at startup rather than on the first message — and finished once when the run loop ends, however it ended, so the ones that hold a part (file has a JSON array to close, s3 has an object that has not been uploaded at all) close it.

Where the destination is a system rather than a place, the settings that describe the system live on a connection and only what this pipeline wants from it — a topic, a table, a path — is on the component. The two exceptions are deliberate: stdout has nothing to connect to, and the http output takes a url, because for a webhook the url is the whole of what a connection would have held.

Two of them map messages onto real columns rather than writing JSON whole; postgres and clickhouse share that mapping and differ only in DDL and wire format. See database outputs.

stdout

Prints each batch to the server's stdout. Useful while building a pipeline up; takes no settings.

This component takes no configuration.

file

Writes each batch to files in a directory on the server.

The directory comes from a file connection and the path below is relative to it; the server's --data-dir is what both are confined to, so a server started without that flag cannot write files at all. Names are generated rather than configured — <open time>-<sequence>.<ext>, which sorts chronologically and cannot collide across rotations.

Meant for local development and testing. The object-store output is what this shape is being built towards for anything else.

field	type		description
connection	file connection	required	name of the file connection to write under — see "connections" in the readme. The root directory lives there; the path below is this output's own.
path	string	required	directory to write into, relative to the connection's root, e.g. orders. Must stay inside the root: an absolute path or one containing .. is refused rather than trimmed.
format	ndjson | json_array	optional	how the messages are laid out. Defaults to ndjson.
rotate	object	optional	when to close a file and start the next one. Without this, one file per run.

rotate

field	type		description
interval_secs	integer	optional	close the file this many seconds after it was opened. Measured from the open, not from the last write, so files line up on a predictable cadence.
max_rows	integer	optional	close the file once it holds this many messages

s3

Writes each batch to objects under a prefix in an S3-compatible bucket.

The same writer as the file output — the same part naming, the same formats, the same rotation policy — pointed at a bucket instead of a directory. What differs is that an object store has no append: a part is buffered in memory and uploaded whole when it rotates, so rotate is required here and is what decides both how often objects appear and how much a running pipeline holds.

field	type		description
connection	s3 connection	required	name of the s3 connection to write through — see "connections" in the readme. The bucket and credentials live there; the prefix below is this output's own.
prefix	string	required	key prefix to write under, e.g. orders — objects land at <prefix>/<generated part name>. Leave it empty to write at the root of the bucket.
rotate	object	required	when to finish an object and start the next one. Required: an object store cannot be appended to, so without a rotation trigger a pipeline would hold its entire run in memory and upload it once, at the end.
format	ndjson | json_array	optional	how the messages are laid out. Defaults to ndjson.

rotate

field	type		description
interval_secs	integer	optional	close the file this many seconds after it was opened. Measured from the open, not from the last write, so files line up on a predictable cadence.
max_rows	integer	optional	close the file once it holds this many messages

kafka

Publishes every message in the batch to a kafka topic, one message per record. Records are sent without a key, so they round-robin across the topic's partitions.

field	type		description
connection	kafka connection	required	name of the kafka connection to publish to — see "connections" in the readme.
topic	string	required	the topic to publish to

nats

Publishes every message in the batch to a nats subject, one message per publish.

field	type		description
connection	nats connection	required	name of the nats connection to publish on — see "connections" in the readme.
subject	string	required	the subject to publish to

postgres

Inserts every message in the batch into a postgres table, one row per message.

With columns, each entry names a column, its type and the field to read — {"name": "temperature", "type": "float", "field": "reading.temp_c"}, and field defaults to the column's name. Without them the table gets a single jsonb column holding the whole message, which is what this output has always done.

The table is created if it isn't there, from the columns above; set create_table to false for a table someone else owns. Creation never alters an existing table — a table whose shape has moved on fails the insert with the server's own error rather than being migrated from a config file.

field	type		description
connection	postgres connection	required	name of the postgres connection to insert through — see "connections" in the readme. The host, database and role live there; the table below is this output's own.
table	string	required	the table to insert into, created if it does not exist. Optionally schema-qualified (analytics.readings); letters, digits and underscores only, since it cannot be sent as a query parameter.
columns	list of column	optional	which message field goes in which column. Leave it out to store each message whole, as JSON, in a payload column.
create_table	boolean	optional	create the table on connect if it does not exist. Defaults to true.
indexes	list of index	optional	indexes to create with the table. Each names mapped columns, in order.
on_extra_fields	ignore | error	optional	what to do about a message carrying fields no column reads
primary_key	list of string	optional	the columns forming the created table's primary key. With none, the table gets an id of its own and a received_at timestamp; naming one here says the data carries its own identity and drops both.

columns — each entry

field	type		description
name	string	required	the column's name in the table. Letters, digits and underscores only, since it cannot be sent as a query parameter.
type	text | integer | bigint | float | decimal | boolean | timestamp | date | uuid | json	required	what the column holds. Values are checked against it rather than coerced into it.
field	string	optional	the field to read, as a dotted path. Defaults to the column's name.
message	boolean	optional	store the whole message in this column instead of one of its fields. Only for a json column, and not together with field.
nullable	boolean	optional	whether the column accepts NULL. Defaults to true; false makes the created column NOT NULL and makes a missing field an error.
on_missing	null | error | skip_row	optional	what to do about a message that doesn't carry the field. Defaults to null, or to error for a column that is not nullable.

indexes — each entry

field	type		description
columns	list of string	required	the columns to index, in order. Each must be one of the mapped columns.
unique	boolean	optional	whether the index is unique. Defaults to false.

clickhouse

Inserts every batch into a ClickHouse table, one insert per batch.

columns is spelled exactly as the postgres output's is — each entry names a column, its type and the field to read, and field defaults to the column's name. Without them the table gets a single column holding each message as JSON text.

Where it differs from postgres is what a created table is sorted by. ClickHouse has no auto-increment column and no unique constraint, so there is no surrogate id to fall back on: order_by names the MergeTree sorting key, and a table that names none is sorted by the received_at timestamp it gets for free. A sorting key does not deduplicate — naming one says how the table is laid out and indexed, not that its rows are unique.

The table is created if it isn't there; set create_table to false for a table someone else owns. Creation never alters an existing table.

field	type		description
connection	clickhouse connection	required	name of the clickhouse connection to insert through — see "connections" in the readme. The url, database and user live there; the table below is this output's own.
table	string	required	the table to insert into, created if it does not exist. Optionally database-qualified (analytics.readings), which overrides the connection's database; letters, digits and underscores only, since it cannot be sent as a query parameter.
columns	list of column	optional	which message field goes in which column. Leave it out to store each message whole, as JSON text, in a payload column.
create_table	boolean	optional	create the table on start if it does not exist. Defaults to true.
on_extra_fields	ignore | error	optional	what to do about a message carrying fields no column reads
order_by	list of string	optional	the columns the created table is sorted by — MergeTree's sorting key, and its index. With none, the table gets a received_at timestamp of its own and is sorted by that. Named columns are made NOT NULL, since a nullable key is not something ClickHouse sorts by.

columns — each entry

field	type		description
name	string	required	the column's name in the table. Letters, digits and underscores only, since it cannot be sent as a query parameter.
type	text | integer | bigint | float | decimal | boolean | timestamp | date | uuid | json	required	what the column holds. Values are checked against it rather than coerced into it.
field	string	optional	the field to read, as a dotted path. Defaults to the column's name.
message	boolean	optional	store the whole message in this column instead of one of its fields. Only for a json column, and not together with field.
nullable	boolean	optional	whether the column accepts NULL. Defaults to true; false makes the created column NOT NULL and makes a missing field an error.
on_missing	null | error | skip_row	optional	what to do about a message that doesn't carry the field. Defaults to null, or to error for a column that is not nullable.

mqtt

Publishes every message in the batch to an mqtt topic, one message per publish.

A stable client id is derived from the pipeline's id and this topic, the same as the mqtt input — not configurable, for the same reason.

field	type		description
connection	mqtt connection	required	name of the mqtt connection to publish on — see "connections" in the readme.
topic	string	required	the topic to publish to
qos	at_most_once | at_least_once | exactly_once	optional	the quality of service to publish with. Defaults to at_most_once.
retain	boolean	optional	ask the broker to keep this as the topic's retained message, handed to every future subscriber immediately on subscribe. Defaults to false.

redis

Publishes every message in the batch to a redis channel, one message per publish.

field	type		description
connection	redis connection	required	name of the redis connection to publish on — see "connections" in the readme.
channel	string	required	the channel to publish to

http

Sends the batch to an http endpoint — the pipeline pushes its results at a webhook or an ingest API rather than at a broker.

The counterpart of the http input, and the sending half of what the http transform does: the transform replaces the batch with the reply, this one is the end of the chain and the reply's body is discarded. What is not discarded is its status — anything but a 2xx fails the batch, which is what makes a webhook that is rejecting the data show up on the card rather than being written off as delivered.

field	type		description
url	string	required	endpoint to send to, e.g. https://example.com/hooks/readings
auth	bearer | header	optional	what this output presents to be allowed to send. Absent — the default — sends no credential at all, which is what an open webhook wants.
body	batch | message	optional	what one request carries. Defaults to batch, which is one request per batch.
timeout_seconds	integer	optional	how long one request may take before it is given up on, in seconds. Defaults to 30. A batch whose request times out is a failed batch, so this is also the longest a slow endpoint can hold the pipeline up.
verb	GET | POST | PUT | PATCH | DELETE	optional	http method. Defaults to POST. GET and DELETE are refused at build time — an output exists to send the messages somewhere, and a method with no body has nowhere to put them.

auth — type: "bearer"

field	type		description
token	string	required	the token. A ${NAME} reference, so the config file holds the name and the secret store holds the value.

auth — type: "header"

field	type		description
name	string	required	the header's name, matched case-insensitively on the way in. On an http input it may not be one of the headers an envelope passes through, since that would write the credential into the messages.
value	string	required	the exact value that header must have. A ${NAME} reference, as above.