outputs

Every output of a pipeline gets every batch. The run loop initializes each output before the first batch. If an output cannot initialize, for example because the database is not available, the run loop tries again with a backoff. The pipeline reads no input until all outputs are ready.

When the run loop ends, it calls finish on each output one time. Two outputs need this step. A file output with json_array closes the array. An s3 output uploads the part that it holds in memory.

When the destination is a system, the settings of the system are on a connection. The component has only what this pipeline wants from the system: a topic, a table, a path. Two outputs have no connection. stdout has nothing to connect to. The http output takes a url, because for a webhook the url is all that a connection holds.

postgres and clickhouse map messages onto real columns. They share the same column mapping and differ only in DDL and wire format. See database outputs.

stdout

Prints each batch to the standard output of the server. Use it to test a pipeline. It has no settings.

This component takes no configuration.

file

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

A file connection gives the root directory, and path is relative to it. The root must be inside the --data-dir of the server. Without that flag, a file output fails to build. The output names each file <open time>-<sequence>.<ext>, so the names sort by time and are unique.

Use the file output for local development and tests.

field	type		description
connection	file connection	required	The name of the file connection to write under. Declare the connection in the connections file. The connection gives the root directory.
path	string	required	The directory to write into, relative to the root of the connection, for example orders. The path must stay inside the root. kayak refuses an absolute path and a path that contains ...
format	ndjson | json_array	optional	The layout of the messages. The default is ndjson.
rotate	object	optional	When to close a file and start the next file. Without this setting, the output writes one file while the pipeline runs.

rotate

field	type		description
interval_secs	integer	optional	Close the file this number of seconds after the output opened it. The time starts when the file opens, not at the last write.
max_rows	integer	optional	Close the file when it contains this number of messages. The output does not divide a batch, so a file can contain more messages.

s3

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

The s3 output uses the same file names, formats and rotation as the file output. An object store cannot append to an object. Thus the output keeps the current object in memory and uploads it when it rotates. rotate is required. It sets how frequently objects appear and how much memory the pipeline uses.

field	type		description
connection	s3 connection	required	The name of the s3 connection to write through. Declare the connection in the connections file. The connection gives the bucket and the credentials.
prefix	string	required	The key prefix to write under, for example orders. The output writes each object to <prefix>/<part name>. Set an empty prefix to write at the root of the bucket.
rotate	object	required	When to close an object and start the next object. Required. A rotate with no trigger fails to build.
format	ndjson | json_array	optional	The layout of the messages. The default is ndjson.

rotate

field	type		description
interval_secs	integer	optional	Close the file this number of seconds after the output opened it. The time starts when the file opens, not at the last write.
max_rows	integer	optional	Close the file when it contains this number of messages. The output does not divide a batch, so a file can contain more messages.

kafka

Publishes each message in the batch to a kafka topic, one record for each message.

The records have no key, so kafka distributes them across the partitions of the topic.

field	type		description
connection	kafka connection	required	The name of the kafka connection to publish to. Declare the connection in the connections file.
topic	string	required	The topic to publish to.

nats

Publishes each message in the batch to a nats subject, one message for each publish.

field	type		description
connection	nats connection	required	The name of the nats connection to publish on. Declare the connection in the connections file.
subject	string	required	The subject to publish to.

postgres

Inserts each message in the batch into a postgres table, one row for each message.

With columns, each entry names a column, its type and the field to read, for example {"name": "temperature", "type": "float", "field": "reading.temp_c"}. The default field is the name of the column. Without columns, the table has an id, a received_at timestamp and a payload column of type jsonb that contains the full message.

The output creates the table if it does not exist. Set create_table to false for a table that another system owns. The output does not change an existing table. If the table does not agree with the columns, the insert fails with the error from postgres.

field	type		description
connection	postgres connection	required	The name of the postgres connection to insert through. Declare the connection in the connections file. The connection gives the host, the database and the role.
table	string	required	The table to insert into. The output creates it if it does not exist. You can add a schema (analytics.readings). Use only letters, digits and underscores.
columns	list of column	optional	The column for each message field. If you do not set it, the output keeps each full message as JSON in a payload column.
create_table	boolean	optional	Create the table on connect if it does not exist. The default is true.
indexes	list of index	optional	The indexes to create with the table. Each index names mapped columns, in sequence.
on_extra_fields	ignore | error	optional	What to do with a message that has fields that no column reads.
primary_key	list of string	optional	The columns of the primary key of the created table. If you do not set it, the table gets an id and a received_at timestamp. If you set it, the table does not get these two columns.

columns — each entry

field	type		description
name	string	required	The name of the column in the table. Use only letters, digits and underscores.
type	text | integer | bigint | float | decimal | boolean | timestamp | date | uuid | json	required	The type of the column. The output checks each value against it.
field	string	optional	The field to read, as a dotted path. The default is the name of the column.
message	boolean	optional	Write the full message to this column. Use it only with a json column. Do not use it with field.
nullable	boolean	optional	Whether the column accepts NULL. The default is true. With false, the output makes the column NOT NULL, and a missing field is an error.
on_missing	null | error | skip_row	optional	What to do with a message that does not have the field. The default is null, or 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 column must be a mapped column.
unique	boolean	optional	Whether the index is unique. The default is false.

clickhouse

Inserts each batch into a ClickHouse table, one insert for each batch.

columns has the same format as on the postgres output. Each entry names a column, its type and the field to read. The default field is the name of the column. Without columns, the table has a payload column that contains each message as JSON text.

ClickHouse has no auto-increment column and no unique constraint. order_by names the sorting key of the MergeTree table. If you do not set it, the table gets a received_at timestamp and is sorted by it. A sorting key does not remove duplicate rows.

The output creates the table if it does not exist. Set create_table to false for a table that another system owns. The output does not change an existing table.

field	type		description
connection	clickhouse connection	required	The name of the clickhouse connection to insert through. Declare the connection in the connections file. The connection gives the url, the database and the user.
table	string	required	The table to insert into. The output creates it if it does not exist. You can add a database (analytics.readings). This database replaces the database of the connection. Use only letters, digits and underscores.
columns	list of column	optional	The column for each message field. If you do not set it, the output keeps each full message as JSON text in a payload column.
create_table	boolean	optional	Create the table on start if it does not exist. The default is true.
on_extra_fields	ignore | error	optional	What to do with a message that has fields that no column reads.
order_by	list of string	optional	The columns that sort the created table. This is the sorting key of the MergeTree table and its index. If you do not set it, the table gets a received_at timestamp and is sorted by it. The output makes these columns NOT NULL, because ClickHouse cannot sort by a nullable key.

columns — each entry

field	type		description
name	string	required	The name of the column in the table. Use only letters, digits and underscores.
type	text | integer | bigint | float | decimal | boolean | timestamp | date | uuid | json	required	The type of the column. The output checks each value against it.
field	string	optional	The field to read, as a dotted path. The default is the name of the column.
message	boolean	optional	Write the full message to this column. Use it only with a json column. Do not use it with field.
nullable	boolean	optional	Whether the column accepts NULL. The default is true. With false, the output makes the column NOT NULL, and a missing field is an error.
on_missing	null | error | skip_row	optional	What to do with a message that does not have the field. The default is null, or error for a column that is not nullable.

mqtt

Publishes each message in the batch to an mqtt topic, one message for each publish.

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 publish on. Declare the connection in the connections file.
topic	string	required	The topic to publish to.
qos	at_most_once | at_least_once | exactly_once	optional	The quality of service for each publish. The default is at_most_once.
retain	boolean	optional	Tell the broker to keep the message as the retained message of the topic. The broker sends it to each new subscriber. The default is false.

redis

Publishes each message in the batch to a redis channel, one message for each publish.

field	type		description
connection	redis connection	required	The name of the redis connection to publish on. Declare the connection in the connections file.
channel	string	required	The channel to publish to.

http

Sends the batch to an http endpoint, for example a webhook or an ingest API.

The output ignores the body of the reply. A status other than 2xx fails the batch. The error contains the reply of the endpoint. Use the http transform if the pipeline needs the reply.

field	type		description
url	string	required	The endpoint to send to, for example https://example.com/hooks/readings.
auth	bearer | header	optional	The credential that the output sends. If you do not set it, the output sends no credential.
body	batch | message	optional	The content of one request. The default is batch, one request for each batch.
timeout_seconds	integer	optional	The maximum time for one request, in s. The default is 30. A request that times out fails the batch. Thus a slow endpoint stops the pipeline for this time at most.
verb	GET | POST | PUT | PATCH | DELETE	optional	The http method. The default is POST. GET and DELETE fail to build, because a request with no body cannot send the messages.

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.

indu

Writes messages into Indu Cloud as streams through POST /ingest/v1/streams. A stream is a series that is not a sensor.

Each message gives one reading for each entry in series. For example, a reducer that sends {machine, oee, availability} with two series entries writes two streams for each machine. Indu creates an unknown stream when the key of the connection has permission to create streams. If Indu does not accept all rows, the batch fails. The error contains the row errors from Indu.

field	type		description
connection	indu connection	required	The name of the indu connection to write through. Declare the connection in the connections file.
series	list of object	required	The streams to write, one reading for each message. At least one entry is required.
at	string	optional	The field that contains the time of the reading, as an RFC 3339 string or as ms since the epoch. If you do not set it, the output uses the time at which it sends the batch. An envelope puts the receive time of an input at _meta.received_at.
timeout_seconds	integer	optional	The maximum time for one request, in s. The default is 30.

series — each entry

field	type		description
stream	string	required	The name of the stream in Indu, for example press-3/oee. The name can contain {field} placeholders that the output fills from the message, for example {machine}/oee. Thus one output can write a stream for each machine. The output skips this series for a message that does not have the field of a placeholder.
value	string	required	The field that contains the value, as a path (oee, stats.mean). The value must be a number. The output skips this series for a message where the value is missing or is not a number. The batch does not fail.
unit	string	optional	The unit that Indu records when it creates the stream, for example %. Indu ignores it when the stream exists.

tidepool

Writes each batch into a Tidepool table, one request for each batch.

The table must exist in the Tidepool project. On start, the output checks the columns against the table. Each mapped column must be in the table and have a type that the mapping can write. Each required column of the table must be written. If the check fails, the start fails.

columns has the same format as on the database outputs. If you do not set it, the output sends each message as a row without changes. Tidepool checks each value and refuses a batch that has a problem.

A refused batch fails. The error gives the problems by row and column. The output tries again when the server is busy (503), when the server cannot be reached and on other 5xx errors. It tries again for up to retry_seconds. Each try uses the same idempotency key, so Tidepool does not write a batch two times.

field	type		description
connection	tidepool connection	required	The name of the tidepool connection to write through. Declare the connection in the connections file.
table	string	required	The table to write into, with the name from the Tidepool project.
columns	list of column	optional	The column for each message field. If you do not set it, the output sends each message as a row without changes.
on_extra_fields	ignore | error	optional	What to do with a message that has fields that no column reads.
retry_seconds	integer	optional	The maximum time to try one batch again while the server is busy or cannot be reached, in s. The default is 30.
timeout_seconds	integer	optional	The maximum time for one request, in s. The default is 30.

columns — each entry

field	type		description
name	string	required	The name of the column in the table. Use only letters, digits and underscores.
type	text | integer | bigint | float | decimal | boolean | timestamp | date | uuid | json	required	The type of the column. The output checks each value against it.
field	string	optional	The field to read, as a dotted path. The default is the name of the column.
message	boolean	optional	Write the full message to this column. Use it only with a json column. Do not use it with field.
nullable	boolean	optional	Whether the column accepts NULL. The default is true. With false, the output makes the column NOT NULL, and a missing field is an error.
on_missing	null | error | skip_row	optional	What to do with a message that does not have the field. The default is null, or error for a column that is not nullable.