Bento

Optionally configure a policy to flush buffered messages in batches.

The maximum buffer size (in bytes) to allow before applying backpressure upstream. Default: 524288000

Whether to drop incoming messages that will exceed the buffer limit. Default: false

The path of the database file, which will be created if it does not already exist.

An optional list of processors to apply to messages after they are consumed from the buffer. These processors are useful for undoing any compression, archiving, etc that may have been done by your pre_processors.

An optional list of processors to apply to messages before they are stored within the buffer. These processors are useful for compressing, archiving or otherwise reducing the data in size before it's stored on disk.

An optional duration string describing the length of time to wait after a window has ended before flushing it, allowing late arrivals to be included. Since this windowing buffer uses the system clock an allowed lateness can improve the matching of messages when using event time. Default:

An optional duration string to offset the beginning of each window by, otherwise they are aligned to the zeroth minute and zeroth hour on the UTC clock. The offset cannot be a larger or equal measure to the window size or the slide. Default:

A duration string describing the size of each window. By default windows are aligned to the zeroth minute and zeroth hour on the UTC clock, meaning windows of 1 hour duration will match the turn of each hour in the day, this can be adjusted with the offset field.

An optional duration string describing by how much time the beginning of each window should be offset from the beginning of the previous, and therefore creates sliding windows instead of tumbling. When specified this duration must be smaller than the size of the window. Default:

A Bloblang mapping applied to each message during ingestion that provides the timestamp to use for allocating it a window. By default the function now() is used in order to generate a fresh timestamp at the time of ingestion (the processing time), whereas this mapping can instead extract a timestamp from the message itself (the event time).

The timestamp value assigned to root must either be a numerical unix time in seconds (with up to nanosecond precision via decimals), or a string in ISO 8601 format. If the mapping fails or provides an invalid result the message will be dropped (with logging to describe the problem). Default: root = now()

Whether to add a failed message payload to an error log. Default: false

Whether to enable message-level error logging. Default: false

Sets the ratio of errored messages within a batch to sample. Default: 1

Encryption algorithm used to generate password_hash. Default: sha256

Enable basic authentication Default: false

Hashed password required to authenticate. (base64 encoded) Default:

Custom realm name Default: restricted

Salt for scrypt algorithm. (base64 encoded) Default:

Username required to authenticate. Default:

Appends additional headers to the list of default allowed headers: Accept, Accept-Language, Content-Language & Origin. These default headers are therefore always allowed. Default: []

Used to explicitly set allowed methods in the Access-Control-Allow-Methods header. Default: [GET HEAD POST PUT PATCH DELETE]

An explicit list of origins that are allowed for CORS requests. Default: []

Whether to allow CORS requests. Default: false

Acknowledge messages automatically as they are consumed rather than waiting for acknowledgments from downstream. This can improve throughput and prevent the pipeline from blocking but at the cost of eliminating delivery guarantees. Default: false

Allows you to passively declare bindings for the target queue.

A consumer tag. Default:

A list of regular expression patterns whereby if a message that has failed to be delivered by Bento has an error that matches it will be dropped (or delivered to a dead-letter queue if one exists). By default failed messages are nacked with requeue enabled. Default: []

The maximum number of pending messages to have consumed at a time. Default: 10

The maximum amount of pending messages measured in bytes to have consumed at a time. Default: 0

An AMQP queue to consume from.

Allows you to passively declare the target queue. If the queue already exists then the declaration passively verifies that they match the target fields.

Custom TLS settings can be used to override system defaults.

A list of URLs to connect to. The first URL to successfully establish a connection will be used until the connection is closed. If an item of the list contains commas it will be expanded into multiple URLs.

Experimental: Azure service bus specific option to renew lock if processing takes more then configured lock time Default: false

Specifies the maximum number of unacknowledged messages the sender can transmit. Once this limit is reached, no more messages will arrive until messages are acknowledged and settled. Default: 64

Read additional message header fields into amqp_* metadata properties. Default: false

Enables SASL authentication.

The source address to consume from.

Custom TLS settings can be used to override system defaults.

A URL to connect to.

A list of URLs to connect to. The first URL to successfully establish a connection will be used until the connection is closed. If an item of the list contains commas it will be expanded into multiple URLs.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

Allows you to configure a batching policy.

The maximum gap between the in flight sequence versus the latest acknowledged sequence at a given time. Increasing this limit enables parallel processing and batching at the output level to work on individual shards. Any given sequence will not be committed unless all messages under that offset are delivered in order to preserve at least once delivery guarantees. Default: 1024

The period of time between each update to the checkpoint table. Default: 5s

Optional manual configuration of AWS credentials to use. More information can be found in this document.

Determines the table used for storing and accessing the latest consumed sequence for shards, and for coordinating balanced consumers of streams.

Allows you to specify a custom endpoint for the AWS API. Default:

The period of time after which a client that has failed to update a shard checkpoint is assumed to be inactive. Default: 30s

The period of time between each attempt to rebalance shards across clients. Default: 30s

The AWS region to target. Default:

Whether to consume from the oldest message when a sequence does not yet exist for the stream. Default: true

One or more Kinesis data streams to consume from. Streams can either be specified by their name or full ARN. Shards of a stream are automatically balanced across consumers by coordinating through the provided DynamoDB table. Multiple comma separated streams can be listed in a single element. Shards are automatically distributed across consumers of a stream by coordinating through the provided DynamoDB table. Alternatively, it's possible to specify an explicit shard to consume from with a colon after the stream name, e.g. foo:0 would consume the shard 0 of the stream foo.

The bucket to consume from. If the field sqs.url is specified this field is optional. Default:

The way in which the bytes of a data source should be converted into discrete messages, codecs are useful for specifying how large files or continuous streams of data might be processed in small chunks rather than loading it all in memory. It's possible to consume lines using a custom delimiter with the delim:x codec, where x is the character sequence custom delimiter. Codecs can be chained with /, for example a gzip compressed CSV file can be consumed with the codec gzip/csv.

Optional manual configuration of AWS credentials to use. More information can be found in this document.

Whether to delete downloaded objects from the bucket once they are processed. Default: false

Allows you to specify a custom endpoint for the AWS API. Default:

Forces the client API to use path style URLs for downloading keys, which is often required when connecting to custom endpoints. Default: false

Default: 1000000

An optional path prefix, if set only objects with the prefix are consumed when walking a bucket. Default:

The AWS region to target. Default:

The scanner by which the stream of bytes consumed will be broken out into individual messages. Scanners are useful for processing large sources of data without holding the entirety of it within memory. For example, the csv scanner allows you to process individual CSV rows without loading the entire CSV file in memory at once. Default: map[to_the_end:map[]]

Consume SQS messages in order to trigger key downloads.

Optional manual configuration of AWS credentials to use. More information can be found in this document.

A map used to send custom HTTP headers alongside each SQS operation to AWS. Default: map[]

Whether to delete the consumed message once it is acked. Disabling allows you to handle the deletion using a different mechanism. Default: true

Allows you to specify a custom endpoint for the AWS API. Default:

The maximum number of messages to return on one poll. Valid values: 1 to 10. Default: 10

The AWS region to target. Default:

Whether to set the visibility timeout of the consumed message to zero once it is nacked. Disabling honors the preset visibility timeout specified for the queue. Default: true

Whether to periodically refresh the visibility timeout of in-flight messages to prevent more-than-once delivery while still processing. Default: true

The SQS URL to consume from.

Whether to set the wait time. Enabling this activates long-polling. Valid values: 0 to 20. Default: 0

The way in which the bytes of a data source should be converted into discrete messages, codecs are useful for specifying how large files or continuous streams of data might be processed in small chunks rather than loading it all in memory. It's possible to consume lines using a custom delimiter with the delim:x codec, where x is the character sequence custom delimiter. Codecs can be chained with /, for example a gzip compressed CSV file can be consumed with the codec gzip/csv.

The name of the container from which to download blobs.

Whether to delete downloaded objects from the blob once they are processed. Default: false

Default: 1000000

An optional path prefix, if set only objects with the prefix are consumed. Default:

The scanner by which the stream of bytes consumed will be broken out into individual messages. Scanners are useful for processing large sources of data without holding the entirety of it within memory. For example, the csv scanner allows you to process individual CSV rows without loading the entire CSV file in memory at once. Default: map[to_the_end:map[]]

The storage account access key. This field is ignored if storage_connection_string is set. Default:

The storage account to access. This field is ignored if storage_connection_string is set. Default:

A storage account connection string. This field is required if storage_account and storage_access_key / storage_sas_token are not set. Default:

The storage account SAS token. This field is ignored if storage_connection_string or storage_access_key are set. Default:

EXPERIMENTAL: An optional source of download targets, configured as a regular Bento input. Each message yielded by this input should be a single structured object containing a field name, which represents the blob to be downloaded.

Account key.

A Bloblang mapping that, for each message, creates a list of arguments to use with the query.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

The maximum number of messages that should be accumulated into each batch. Use '-1' specify dynamic page size. Default: -1

Connection string.

Container.

Database.

CosmosDB endpoint.

A Bloblang mapping which should evaluate to a single partition key value or an array of partition key values of type string, integer or boolean. Currently, hierarchical partition keys are not supported so only one value may be provided.

The query to execute

The timeout duration until a dequeued message gets visible again, 30s by default Default: 30s

The maximum number of unprocessed messages to fetch at a given time. Default: 10

The name of the source storage queue.

The storage account access key. This field is ignored if storage_connection_string is set. Default:

The storage account to access. This field is ignored if storage_connection_string is set. Default:

A storage account connection string. This field is required if storage_account and storage_access_key / storage_sas_token are not set. Default:

Default:

If set to true the queue is polled on each read request for information such as the queue message lag. These properties are added to consumed messages as metadata, but will also have a negative performance impact. Default: false

Whether to automatically acknowledge messages as they are consumed rather than waiting for acknowledgments from downstream. This can improve throughput but at the cost of eliminating delivery guarantees. Default: false

The Service Bus connection string. This can be obtained from the Azure portal. If not provided, namespace and default credentials will be used. Default:

The maximum number of unprocessed messages to fetch at a given time. Default: 10

A list of regular expression patterns whereby if a message that has failed to be delivered by Bento has an error that matches it will be nacked (or sent to dead letter queue if configured). By default failed messages are nacked with requeue enabled. Default: []

The Service Bus namespace. Required when connection_string is not provided. Default:

The name of the Service Bus queue to consume from.

Automatically renew message locks to prevent lock expiration during processing. Useful for long-running message processing. Default: true

OData filter expression. Is not set all rows are returned. Valid operators are eq, ne, gt, lt, ge and le Default:

Maximum number of records to return on each page. Default: 1000

Select expression using OData notation. Limits the columns on each record to just those requested. Default:

The storage account access key. This field is ignored if storage_connection_string is set. Default:

The storage account to access. This field is ignored if storage_connection_string is set. Default:

A storage account connection string. This field is required if storage_account and storage_access_key / storage_sas_token are not set. Default:

The storage account SAS token. This field is ignored if storage_connection_string or storage_access_key are set. Default:

The table to read messages from.

The child input.

Allows you to configure a batching policy.

An address to connect to.

Allows you to configure a batching policy.

Whatever is specified within inputs will be created this many times. Default: 1

A list of inputs to create.

A list of Cassandra nodes to connect to. Multiple comma separated addresses can be specified on a single line.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

Control time intervals between retry attempts.

If enabled the driver will not attempt to get host info from the system.peers table. This can speed up queries but will mean that data_centre, rack and token information will not be available. Default: false

The maximum number of retries before giving up on a request. Default: 3

Optional configuration of Cassandra authentication parameters.

A query to execute.

The client connection timeout. Default: 600ms

Custom TLS settings can be used to override system defaults.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

A cache resource to use for storing the current latest cursor that has been successfully delivered, this allows Bento to continue from that cursor upon restart, rather than consume the entire state of the table.

A Data Source Name to identify the target database.

A list of options to be included in the changefeed (WITH X, Y...). NOTE: Both the CURSOR option and UPDATED will be ignored from these options when a cursor_cache is specified, as they are set explicitly by Bento in this case.

CSV of tables to be included in the changefeed

Custom TLS settings can be used to override system defaults.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

Optionally process records in batches. This can help to speed up the consumption of exceptionally large CSV files. When the end of the file is reached the remaining records are processed as a (potentially smaller) batch. Default: 1

Whether to delete input files from the disk once they are fully consumed. Default: false

The delimiter to use for splitting values in each record. It must be a single character. Default: ,

An optional list of expected headers in the header row. If provided, the scanner will check the file contents and emit an error if any expected headers don't match.

The number of expected fields in the csv file.

If set to true, a quote may appear in an unquoted field and a non-doubled quote may appear in a quoted field. Default: false

Whether to reference the first row as a header row. If set to true the output structure for messages will be an object where field keys are determined by the header row. Otherwise, each message will consist of an array of values from the corresponding CSV row. Default: true

A list of file paths to read from. Each file will be read sequentially until the list is exhausted, at which point the input will close. Glob patterns are supported, including super globs (double star).

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

Basic Authentication fields

The name of the database to connect to.

Set to true to connect without authentication. Default: false

The cypher query to execute.

Custom TLS settings can be used to override system defaults.

The URL of the database engine.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

A bot token used for authentication.

A cache resource to use for performing unread message backfills, the ID of the last message received will be stored in this cache and used for subsequent requests.

The key identifier used when storing the ID of the last message received. Default: last_message_id

A discord channel ID to consume messages from.

The maximum number of messages to receive in a single request. Default: 100

The length of time (as a duration string) to wait between each poll for backlogged messages. This field can be set empty, in which case requests are made at the limit set by the rate limit. This field also supports cron expressions. Default: 1m

Default: An optional rate limit resource to restrict API requests with.

A map of inputs to statically create. Default: map[]

A path prefix for HTTP endpoints that are registered. Default:

Optional configuration of etcd authentication headers.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

The interval to update endpoints with its latest members. 0 disables auto-sync. By default auto-sync is disabled.

The jitter fraction to randomize backoff wait time.

The wait time before retrying an RPC.

Timeout for failing to establish a connection. Default: 5s

A set of URLs (schemes, hosts and ports only) that can be used to communicate with a logical etcd cluster. If multiple endpoints are provided, the Client will attempt to use them all in the event that one or more of them are unusable.

Time after which client pings the server to see if transport is alive. Default: 5s

Time that the client waits for a response for the keep-alive probe. If the response is not received in this time, the connection is closed. Default: 1s

The key or prefix being watched. For prefix watching, options.with_prefix should be true

The client-side response receive limit. If 0, it defaults to math.MaxInt32.

The client-side request send limit in bytes. If 0, it defaults to 2.0 MiB (2 * 1024 * 1024).

The maximum number of retries for unary RPCs.

Collection of options to configure an etcd watcher.

When set, will allow client to send keepalive pings to server without any active streams (RPCs). Default: false

When set, will refuse to create a client against an outdated cluster. Default: false

Timeout for a single request. This includes connection time, any redirects, and header wait time. Default: 1s

Custom TLS settings can be used to override system defaults.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

The way in which the bytes of a data source should be converted into discrete messages, codecs are useful for specifying how large files or continuous streams of data might be processed in small chunks rather than loading it all in memory. It's possible to consume lines using a custom delimiter with the delim:x codec, where x is the character sequence custom delimiter. Codecs can be chained with /, for example a gzip compressed CSV file can be consumed with the codec gzip/csv.

Whether to delete input files from the disk once they are fully consumed. Default: false

Default: 1000000

A list of paths to consume sequentially. Glob patterns are supported, including super globs (double star).

The scanner by which the stream of bytes consumed will be broken out into individual messages. Scanners are useful for processing large sources of data without holding the entirety of it within memory. For example, the csv scanner allows you to process individual CSV rows without loading the entire CSV file in memory at once. Default: map[lines:map[]]

An optional Bloblang mapping which should evaluate to an array of values matching in size to the number of placeholder arguments in the field where.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

A list of columns to query.

A list of labels to add to the query job. Default: map[]

An optional prefix to prepend to the select query (before SELECT).

The priority with which to schedule the query. Default:

GCP project where the query job will execute.

An optional suffix to append to the select query.

Fully-qualified BigQuery table name to query.

An optional where clause to add. Placeholder arguments are populated with the args_mapping field. Placeholders should always be question marks (?).

The name of the bucket from which to download objects.

The way in which the bytes of a data source should be converted into discrete messages, codecs are useful for specifying how large files or continuous streams of data might be processed in small chunks rather than loading it all in memory. It's possible to consume lines using a custom delimiter with the delim:x codec, where x is the character sequence custom delimiter. Codecs can be chained with /, for example a gzip compressed CSV file can be consumed with the codec gzip/csv.

Whether to delete downloaded objects from the bucket once they are processed. Default: false

Default: 1000000

An optional path prefix, if set only objects with the prefix are consumed. Default:

The scanner by which the stream of bytes consumed will be broken out into individual messages. Scanners are useful for processing large sources of data without holding the entirety of it within memory. For example, the csv scanner allows you to process individual CSV rows without loading the entire CSV file in memory at once. Default: map[to_the_end:map[]]

Allows you to configure the input subscription and creates if it doesn't exist.

An optional endpoint to override the default of pubsub.googleapis.com:443. This can be used to connect to a region specific pubsub endpoint. For a list of valid values check out this document. Default:

The maximum number of outstanding pending messages to be consumed measured in bytes. Default: 1e+09

The maximum number of outstanding pending messages to be consumed at a given time. Default: 1000

The project ID of the target subscription.

The target subscription ID.

Enable synchronous pull mode. Default: false

An optional field to define the end time to read from the changestreams, timestamp format should conform to RFC3339

An optional field to configure the heartbeat interval for partitions. Default: 3s

The maximum number of messages to have pulled in. Default: 1024

The dsn for spanner from where to read the changestream.

An optional field to define the start point to read from the changestreams. If not set then the current time is used. The timestamp format should conform to RFC3339, for details on valid start times please see this document

The name of the stream to track changes on.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

The number of generated messages that should be accumulated into each batch flushed at the specified interval. Default: 1

An optional number of messages to generate, if set above 0 the specified number of messages is generated and then the input will shut down. Default: 0

The time interval at which messages should be generated, expressed either as a duration string or as a cron expression. If set to an empty string messages will be generated as fast as downstream services can process them. Cron expressions can specify a timezone by prefixing the expression with TZ=<location name>, where the location name corresponds to a file within the IANA Time Zone database. Default: 1s

A bloblang mapping to use for generating messages.

The directory to consume from.

A list of target host addresses to connect to.

A user ID to connect as. Default:

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

A list of status codes whereby the request should be considered to have failed and retries should be attempted, but the period between them should be increased gradually. Default: [429]

Allows you to specify basic authentication.

Whether empty payloads received from the target server should be dropped. Default: true

A list of status codes whereby the request should be considered to have failed but retries should not be attempted. This is useful for preventing wasted retries for requests that will never succeed. Note that with these status codes the request is dropped, but message that caused the request will not be dropped. Default: []

EXPERIMENTAL: Optionally set a level at which the request and response payload of each request made will be logged. Default:

Specify which response headers should be added to resulting messages as metadata. Header keys are lowercased before matching, so ensure that your patterns target lowercased versions of the header keys that you expect.

A map of headers to add to the request. Default: map[]

BETA: Allows you to specify JWT authentication.

The maximum period to wait between failed requests. Default: 300s

Specify optional matching rules to determine which metadata keys should be added to the HTTP request as headers.

Allows you to specify open authentication via OAuth version 1.

Allows you to specify open authentication via OAuth version 2 using the client credentials token flow.

An optional payload to deliver for each request.

An optional HTTP proxy URL.

An optional rate limit to throttle requests by.

The maximum number of retry attempts to make. Default: 3

The base period to wait between failed requests. Default: 1s

Allows you to set streaming mode, where requests are kept open and messages are processed line-by-line.

A list of status codes whereby the attempt should be considered successful, this is useful for dropping requests that return non-2XX codes indicating that the message has been dealt with, such as a 303 See Other or a 409 Conflict. All 2XX codes are considered successful unless they are present within backoff_on or drop_on, regardless of this field. Default: []

A static timeout to apply to requests. Default: 5s

Custom TLS settings can be used to override system defaults.

Custom transport options.

The URL to connect to.

A verb to connect with Default: GET

An alternative address to host from. If left empty the service wide address is used. Default:

An array of verbs that are allowed for the path endpoint. Default: [POST]

Enable TLS by specifying a certificate and key file. Only valid with a custom address. Default:

Adds Cross-Origin Resource Sharing headers. Only valid with a custom address.

Enable TLS by specifying a certificate and key file. Only valid with a custom address. Default:

The endpoint path to listen for POST requests. Default: /post

An optional rate limit to throttle requests by. Default:

Customise messages returned via synchronous responses.

Timeout for requests. If a consumed messages takes longer than this to be delivered the connection is closed, but the message may still be delivered. Default: 5s

The endpoint path to create websocket connections from. Default: /post/ws

An optional message to delivery to websocket connections that are rate limited. Default:

An optional message to deliver to fresh websocket connections. Default:

A list of broker addresses to connect to. If an item of the list contains commas it will be expanded into multiple addresses.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

Allows you to configure a batching policy.

The maximum number of messages of the same topic and partition that can be processed at a given time. Increasing this limit enables parallel processing and batching at the output level to work on individual partitions. Any given offset will not be committed unless all messages under that offset are delivered in order to preserve at least once delivery guarantees. Default: 1024

An identifier for the client connection. Default: bento

The period of time between each commit of the current partition offsets. Offsets are always committed during shutdown. Default: 1s

An identifier for the consumer group of the connection. This field can be explicitly made empty in order to disable stored offsets for the consumed topic partitions. Default:

EXPERIMENTAL: A Bloblang mapping that attempts to extract an object containing tracing propagation information, which will then be used as the root tracing span for the message. The specification of the extracted fields must match the format used by the service wide tracer.

The maximum number of unprocessed messages to fetch at a given time. Default: 256

Tuning parameters for consumer group synchronization.

A maximum estimate for the time taken to process a message, this is used for tuning consumer group synchronization. Default: 100ms

Decode headers into lists to allow handling of multiple values with the same key Default: false

EXPERIMENTAL: Starts a new root span with link to parent.

A rack identifier for this client. Default:

Enables SASL authentication.

Determines whether to consume from the oldest available offset, otherwise messages are consumed from the latest offset. The setting is applied when creating a new consumer group or the saved offset no longer exists. Default: true

The version of the Kafka protocol to use. This limits the capabilities used by the client and should ideally match the version of your brokers. Defaults to the oldest supported stable version.

Custom TLS settings can be used to override system defaults.

A list of topics to consume from. Multiple comma separated topics can be listed in a single element. Partitions are automatically distributed across consumers of a topic. Alternatively, it's possible to specify explicit partitions to consume from with a colon after the topic name, e.g. foo:0 would consume the partition 0 of the topic foo. This syntax supports ranges, e.g. foo:0-10 would consume partitions 0 through to 10 inclusive.

Determines which offset to automatically consume from, matching Kafka's auto.offset.reset property. When specified, this takes precedence over start_from_oldest.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

Allows you to configure a batching policy that applies to individual topic partitions in order to batch messages together before flushing them for processing. Batching can be beneficial for performance as well as useful for windowed processing, and doing so this way preserves the ordering of topic partitions.

:::caution Setting this checkpoint_limit: 1will not enforce 'strict ordered' processing of records. Use the kafka input processor for 'strict ordered' processing. :::

		Determines how many messages of the same partition can be processed in parallel before applying back pressure. When a message of a given offset is delivered to the output the offset is only allowed to be committed when all messages of prior offsets have also been delivered, this ensures at-least-once delivery guarantees. However, this mechanism also increases the likelihood of duplicates in the event of crashes or server faults, reducing the checkpoint limit will mitigate this. Default: 1024

An identifier for the client connection. Default: bento

The period of time between each commit of the current partition offsets. Offsets are always committed during shutdown. Default: 5s

An optional consumer group to consume as. When specified the partitions of specified topics are automatically distributed across consumers sharing a consumer group, and partition offsets are automatically committed and resumed under this name. Consumer groups are not supported when specifying explicit partitions to consume from in the topics field.

This sets the maximum amount of bytes a broker will try to send during a fetch. Note that brokers may not obey this limit if it has records larger than this limit. Also note that this client sends a fetch to each broker concurrently, meaning the client will buffer up to <brokers * max bytes> worth of memory. Equivalent to Kafka's fetch.max.bytes option. Default: 50MiB

Sets the maximum amount of bytes that will be consumed for a single partition in a fetch request. Note that if a single batch is larger than this number, that batch will still be returned so the client can make progress. Equivalent to Kafka's max.partition.fetch.bytes option. Default: 1MiB

This sets the maximum amount of time a broker will wait for a fetch response to hit the minimum number of required bytes before returning, overriding the default 5s. Default: 5s

Balancers sets the group balancers to use for dividing topic partitions among group members. This option is equivalent to Kafka's partition.assignment.strategies option. Default: [cooperative_sticky]

This sets the maximum age for the client's cached metadata, to allow detection of new topics, partitions, etc. Default: 5m

Decode headers into lists to allow handling of multiple values with the same key Default: false

This allows you to re-order partitions before they are fetched, given each partition's current lag.

By default, the client rotates partitions fetched by one after every fetch request. Kafka answers fetch requests in the order that partitions are requested, filling the fetch response untilfetch_max_bytes and fetch_max_partition_bytes are hit. All partitions eventually rotate to the front, ensuring no partition is starved.

With this option, you can return topic order and per-topic partition ordering. These orders will sort to the front (first by topic, then by partition). Any topic or partitions that you do not return are added to the end, preserving their original ordering.

A rack identifier for this client. Default:

An optional rate_limit to throttle invocations by. Default:

Determines whether to close the client and force a reconnect after seeing an UNKNOWN_TOPIC_OR_PARTITION or UNKNOWN_TOPIC_ID error. Default: false

Whether listed topics should be interpreted as regular expression patterns for matching multiple topics. When topics are specified with explicit partitions this field must remain set to false. Default: false

Specify one or more methods of SASL authentication. SASL is tried in order; if the broker supports the first mechanism, all connections will use that mechanism. If the first mechanism fails, the client will pick the first supported mechanism. If the broker does not support any client mechanisms, connections will fail.

A list of broker addresses to connect to in order to establish connections. If an item of the list contains commas it will be expanded into multiple addresses.

Determines whether to consume from the oldest available offset, otherwise messages are consumed from the latest offset. The setting is applied when creating a new consumer group or the saved offset no longer exists. Default: true

Custom TLS settings can be used to override system defaults.

A list of topics to consume from. Multiple comma separated topics can be listed in a single element. When a consumer_group is specified partitions are automatically distributed across consumers of a topic, otherwise all partitions are consumed.

Alternatively, it's possible to specify explicit partitions to consume from with a colon after the topic name, e.g. foo:0 would consume the partition 0 of the topic foo. This syntax supports ranges, e.g. foo:0-10 would consume partitions 0 through to 10 inclusive.

Finally, it's also possible to specify an explicit offset to consume from by adding another colon after the partition, e.g. foo:0:10 would consume the partition 0 of the topic foo starting from the offset 10. If the offset is not present (or remains unspecified) then the field start_from_oldest determines which offset to start from.

Kubernetes API server URL. Only used when providing explicit credentials. Default:

Automatically detect authentication method. Tries in-cluster config first, then falls back to kubeconfig. Default: true

Control time intervals between retry attempts.

Path to CA certificate file for verifying API server. Default:

Burst limit for Kubernetes API client. 0 uses the client-go default. Default: 0

QPS limit for Kubernetes API client. 0 uses the client-go default. Default: 0

Kubernetes context to use from kubeconfig. If empty, uses the current context. Default:

Custom Resource Definition to watch. Use this for CRDs instead of 'resource'.

Watch event types to include. Valid values are ADDED, MODIFIED, and DELETED. Default: [ADDED MODIFIED DELETED]

Kubernetes field selector to filter resources. Default: map[]

Emit ADDED events for all existing resources when starting. Default: true

Skip TLS certificate verification. Not recommended for production. Default: false

Path to kubeconfig file. If empty, uses $KUBECONFIG (if set) or the default kubeconfig location (~/.kube/config). Default:

Kubeconfig content as a YAML string. Use this to embed kubeconfig directly in your config (e.g., from a secret or environment variable) instead of referencing a file path. Default:

Kubernetes label selector to filter resources. Default: map[]

The maximum number of retries before giving up on the request. If set to zero there is no discrete limit. Default: 0

Namespaces to watch. Empty list means all namespaces. Default: []

Timeout for Kubernetes API requests such as list calls. Use "0s" to disable. Default: 30s

Standard Kubernetes resource type to watch. Default:

Bearer token for authentication. Can be a service account token. Default:

Path to file containing bearer token. Default:

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

A explicit number of documents to batch up before flushing them for processing. Must be greater than 0. Operations: find, aggregate

The collection to select from.

The name of the target MongoDB database.

The json_marshal_mode setting is optional and controls the format of the output message. Default: canonical

An explicit maximum number of documents to return. Operations: find

The mongodb operation to perform. Default: find

The password to connect to the database. Default:

Bloblang expression describing MongoDB query.

An object specifying fields to sort by, and the respective sort order (1 ascending, -1 descending). Note: The driver currently appears to support only one sorting key. Operations: find

The URL of the target MongoDB server.

The username to connect to the database. Default:

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

Set whether the connection is non-persistent. Default: true

An identifier for the client connection. Default:

The maximum amount of time to wait in order to establish a connection before the attempt is abandoned. Default: 30s

Append a dynamically generated suffix to the specified client_id on each run of the pipeline. This can be useful when clustering Bento producers.

Max seconds of inactivity before a keepalive message is sent. Default: 30

A password to connect with. Default:

The level of delivery guarantee to enforce. Has options 0, 1, 2. Default: 1

Custom TLS settings can be used to override system defaults.

A list of topics to consume from.

A list of URLs to connect to. If an item of the list contains commas it will be expanded into multiple URLs.

A username to connect with. Default:

Set last will message in case of Bento failure

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

Whether the URLs provided should be connected to, or bound as. Default: true

The period to wait until a poll is abandoned and reattempted. Default: 5s

The socket type to use. Default: PULL

A list of subscription topic filters to use when consuming from a SUB socket. Specifying a single sub_filter of '' will subscribe to everything. Default: []

A list of URLs to connect to (or as). If an item of the list contains commas it will be expanded into multiple URLs.

Optional configuration of NATS authentication parameters.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

EXPERIMENTAL: A Bloblang mapping that attempts to extract an object containing tracing propagation information, which will then be used as the root tracing span for the message. The specification of the extracted fields must match the format used by the service wide tracer.

An optional delay duration on redelivering a message when negatively acknowledged.

EXPERIMENTAL: Starts a new root span with link to parent.

The maximum number of messages to pull at a time. Default: 524288

An optional queue group to consume as.

A subject to consume from. Supports wildcards for consuming multiple subjects. Either a subject or stream must be specified.

Custom TLS settings can be used to override system defaults.

A list of URLs to connect to. If an item of the list contains commas it will be expanded into multiple URLs.

The maximum amount of time NATS server should wait for an ack from consumer. Default: 30s

Optional configuration of NATS authentication parameters.

Indicates that the subscription should use an existing consumer.

Determines which messages to deliver when consuming without a durable subscriber. Default: all

Preserve the state of your consumer under a durable name.

EXPERIMENTAL: A Bloblang mapping that attempts to extract an object containing tracing propagation information, which will then be used as the root tracing span for the message. The specification of the extracted fields must match the format used by the service wide tracer.

The maximum number of outstanding acks to be allowed before consuming is halted. Default: 1024

EXPERIMENTAL: Starts a new root span with link to parent.

An optional queue group to consume as.

A stream to consume from. Either a subject or stream must be specified.

A subject to consume from. Supports wildcards for consuming multiple subjects. Either a subject or stream must be specified.

Custom TLS settings can be used to override system defaults.

A list of URLs to connect to. If an item of the list contains commas it will be expanded into multiple URLs.

Optional configuration of NATS authentication parameters.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

The name of the KV bucket.

Do not send delete markers as messages. Default: false

Include all the history per key, not just the last one. Default: false

Key to watch for updates, can include wildcards. Default: >

Retrieve only the metadata of the entry Default: false

Custom TLS settings can be used to override system defaults.

A list of URLs to connect to. If an item of the list contains commas it will be expanded into multiple URLs.

Optional configuration of NATS authentication parameters.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

The name of the object store bucket.

Custom TLS settings can be used to override system defaults.

A list of URLs to connect to. If an item of the list contains commas it will be expanded into multiple URLs.

An optional duration to specify at which a message that is yet to be acked will be automatically retried. Default: 30s

Optional configuration of NATS authentication parameters.

A client ID to connect as. Default:

The ID of the cluster to consume from.

Preserve the state of your consumer under a durable name. Default:

EXPERIMENTAL: A Bloblang mapping that attempts to extract an object containing tracing propagation information, which will then be used as the root tracing span for the message. The specification of the extracted fields must match the format used by the service wide tracer.

The maximum number of unprocessed messages to fetch at a given time. Default: 1024

EXPERIMENTAL: Starts a new root span with link to parent.

The queue to consume from. Default:

If a position is not found for a queue, determines whether to consume from the oldest available message, otherwise messages are consumed from the latest. Default: true

A subject to consume from. Default:

Custom TLS settings can be used to override system defaults.

Whether the subscription should be destroyed when this client disconnects. Default: false

A list of URLs to connect to. If an item of the list contains commas it will be expanded into multiple URLs.

An optional secret for NSQ authentication (requires nsqd 0.2.29+).

The channel to consume from.

A list of nsqlookupd addresses to connect to.

The maximum number of attempts to successfully consume a messages. Default: 5

The maximum number of pending messages to consume at any given time. Default: 100

A list of nsqd addresses to connect to.

Custom TLS settings can be used to override system defaults.

The topic to consume from.

A user agent to assume when connecting.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

Optionally process records in batches. This can help to speed up the consumption of exceptionally large files. When the end of the file is reached the remaining records are processed as a (potentially smaller) batch. Default: 1

A list of file paths to read from. Each file will be read sequentially until the list is exhausted, at which point the input will close. Glob patterns are supported, including super globs (double star).

Whether to enforce strict Parquet schema validation. When set to false, allows reading files with non-standard schema structures (such as non-standard LIST formats). Disabling strict mode may reduce validation but increases compatibility. Default: true

Optional configuration of Pulsar authentication methods.

Specify the subscription name for this consumer.

Specify the subscription type for this consumer.

NOTE: Using a key_shared subscription type will allow out-of-order delivery since nack-ing messages sets non-zero nack delivery delay - this can potentially cause consumers to stall. See Pulsar documentation and this Github issue for more details. Default: shared

Specify the path to a custom CA certificate to trust broker TLS service.

A list of topics to subscribe to. This or topics_pattern must be set.

A regular expression matching the topics to subscribe to. This or topics must be set.

A URL to connect to.

A Bloblang query that should return a boolean value indicating whether the input should now be closed.

The maximum amount of time without receiving new messages after which the input is closed.

The child input to consume from.

Whether the input should be reopened if it closes itself before the condition has resolved to true. Default: false

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

The command used to pop elements from the Redis list Default: blpop

The key of a list to read from.

Specifies a simple, cluster-aware, or failover-aware redis client. Default: simple

Name of the redis master when kind is failover Default:

Optionally sets a limit on the number of messages that can be flowing through a Bento stream pending acknowledgment from the input at any given time. Once a message has been either acknowledged or rejected (nacked) it is no longer considered pending. If the input produces logical batches then each batch is considered a single count against the maximum. WARNING: Batching policies at the output level will stall if this field limits the number of messages below the batching threshold. Zero (default) or lower implies no limit. Default: 0

The length of time to poll for new messages before reattempting. Default: 5s

Custom TLS settings can be used to override system defaults.

Troubleshooting

Some cloud hosted instances of Redis (such as Azure Cache) might need some hand holding in order to establish stable connections. Unfortunately, it is often the case that TLS issues will manifest as generic error messages such as "i/o timeout". If you're using TLS and are seeing connectivity problems consider setting enable_renegotiation to true, and ensuring that the server supports at least TLS version 1.2.

The URL of the target Redis server. Database is optional and is supplied as the URL path.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

A list of channels to consume from.

Specifies a simple, cluster-aware, or failover-aware redis client. Default: simple

Name of the redis master when kind is failover Default:

Custom TLS settings can be used to override system defaults.

Troubleshooting

Some cloud hosted instances of Redis (such as Azure Cache) might need some hand holding in order to establish stable connections. Unfortunately, it is often the case that TLS issues will manifest as generic error messages such as "i/o timeout". If you're using TLS and are seeing connectivity problems consider setting enable_renegotiation to true, and ensuring that the server supports at least TLS version 1.2.

The URL of the target Redis server. Database is optional and is supplied as the URL path.

Whether to use the PSUBSCRIBE command, allowing for glob-style patterns within target channel names. Default: false

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

Specifies a simple, cluster-aware, or failover-aware redis client. Default: simple

Name of the redis master when kind is failover Default:

Iterates only elements matching the optional glob-style pattern. By default, it matches all elements. Default:

Custom TLS settings can be used to override system defaults.

Troubleshooting

Some cloud hosted instances of Redis (such as Azure Cache) might need some hand holding in order to establish stable connections. Unfortunately, it is often the case that TLS issues will manifest as generic error messages such as "i/o timeout". If you're using TLS and are seeing connectivity problems consider setting enable_renegotiation to true, and ensuring that the server supports at least TLS version 1.2.

The URL of the target Redis server. Database is optional and is supplied as the URL path.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

The field key to extract the raw message from. All other keys will be stored in the message as metadata. Default: body

An identifier for the client connection. Default:

The period of time between each commit of the current offset. Offsets are always committed during shutdown. Default: 1s

An identifier for the consumer group of the stream. Default:

Create subscribed streams if they do not exist (MKSTREAM option). Default: true

Specifies a simple, cluster-aware, or failover-aware redis client. Default: simple

The maximum number of messages to consume from a single request. Default: 10

Name of the redis master when kind is failover Default:

If an offset is not found for a stream, determines whether to consume from the oldest available offset, otherwise messages are consumed from the latest offset. Default: true

A list of streams to consume from.

The length of time to poll for new messages before reattempting. Default: 1s

Custom TLS settings can be used to override system defaults.

Troubleshooting

Some cloud hosted instances of Redis (such as Azure Cache) might need some hand holding in order to establish stable connections. Unfortunately, it is often the case that TLS issues will manifest as generic error messages such as "i/o timeout". If you're using TLS and are seeing connectivity problems consider setting enable_renegotiation to true, and ensuring that the server supports at least TLS version 1.2.

The URL of the target Redis server. Database is optional and is supplied as the URL path.

Access token for S2 account

Interval to backoff for before reconnecting to a stream Default: 100ms

Basin name

Cache resource label for storing sequence number

Optionally sets a limit on the number of messages that can be flowing through a Bento stream pending acknowledgment from the input at any given time. Once a message has been either acknowledged or rejected (nacked) it is no longer considered pending. If the input produces logical batches then each batch is considered a single count against the maximum. WARNING: Batching policies at the output level will stall if this field limits the number of messages below the batching threshold. Zero (default) or lower implies no limit. Default: 0

Start consuming the stream from either the earliest or the latest sequence number Default: earliest

Streams prefix or list of streams to subscribe to

Interval after which the streams list should update dynamically Default: 1m

An array of inputs to read from sequentially.

EXPERIMENTAL: Provides a way to perform outer joins of arbitrarily structured and unordered data resulting from the input sequence, even when the overall size of the data surpasses the memory available on the machine.

When configured the sequence of inputs will be consumed one or more times according to the number of iterations, and when more than one iteration is specified each iteration will process an entirely different set of messages by sharding them by the ID field. Increasing the number of iterations reduces the memory consumption at the cost of needing to fully parse the data each time.

Each message must be structured (JSON or otherwise processed into a structured form) and the fields will be aggregated with those of other messages sharing the ID. At the end of each iteration the joined messages are flushed downstream before the next iteration begins, hence keeping memory usage limited.

The address of the server to connect to.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

The way in which the bytes of a data source should be converted into discrete messages, codecs are useful for specifying how large files or continuous streams of data might be processed in small chunks rather than loading it all in memory. It's possible to consume lines using a custom delimiter with the delim:x codec, where x is the character sequence custom delimiter. Codecs can be chained with /, for example a gzip compressed CSV file can be consumed with the codec gzip/csv.

The credentials to use to log into the target server.

Whether to delete files from the server once they are processed. Default: false

Default: 1000000

A list of paths to consume sequentially. Glob patterns are supported.

The scanner by which the stream of bytes consumed will be broken out into individual messages. Scanners are useful for processing large sources of data without holding the entirety of it within memory. For example, the csv scanner allows you to process individual CSV rows without loading the entire CSV file in memory at once. Default: map[to_the_end:map[]]

An experimental mode whereby the input will periodically scan the target paths for new files and consume them, when all files are consumed the input will continue polling for new files.

The address to connect to.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

The way in which the bytes of a data source should be converted into discrete messages, codecs are useful for specifying how large files or continuous streams of data might be processed in small chunks rather than loading it all in memory. It's possible to consume lines using a custom delimiter with the delim:x codec, where x is the character sequence custom delimiter. Codecs can be chained with /, for example a gzip compressed CSV file can be consumed with the codec gzip/csv.

Default: 1000000

A network type to assume (unix|tcp).

The scanner by which the stream of bytes consumed will be broken out into individual messages. Scanners are useful for processing large sources of data without holding the entirety of it within memory. For example, the csv scanner allows you to process individual CSV rows without loading the entire CSV file in memory at once. Default: map[lines:map[]]

The address to listen from.

An optional cache within which this input should write it's bound address once known. The key of the cache item containing the address will be the label of the component suffixed with _address (e.g. foo_address), or socket_server_address when a label has not been provided. This is useful in situations where the address is dynamically allocated by the server (127.0.0.1:0) and you want to store the allocated address somewhere for reference by other systems and components.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

The way in which the bytes of a data source should be converted into discrete messages, codecs are useful for specifying how large files or continuous streams of data might be processed in small chunks rather than loading it all in memory. It's possible to consume lines using a custom delimiter with the delim:x codec, where x is the character sequence custom delimiter. Codecs can be chained with /, for example a gzip compressed CSV file can be consumed with the codec gzip/csv.

Default: 1000000

A network type to accept.

The scanner by which the stream of bytes consumed will be broken out into individual messages. Scanners are useful for processing large sources of data without holding the entirety of it within memory. For example, the csv scanner allows you to process individual CSV rows without loading the entire CSV file in memory at once. Default: map[lines:map[]]

TLS specific configuration, valid when the network is set to tls.

A Bloblang mapping which should evaluate to an array of values matching in size to the number of columns specified.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

Optional Fields that can be set to use Azure based authentication for Azure Postgres SQL

An optional maximum number of connections in the idle connection pool. If conn_max_open is greater than 0 but less than the new conn_max_idle, then the new conn_max_idle will be reduced to match the conn_max_open limit. If value <= 0, no idle connections are retained. The default max idle connections is currently 2. This may change in a future release. Default: 2

An optional maximum amount of time a connection may be idle. Expired connections may be closed lazily before reuse. If value <= 0, connections are not closed due to a connections idle time.

An optional maximum amount of time a connection may be reused. Expired connections may be closed lazily before reuse. If value <= 0, connections are not closed due to a connections age.

An optional maximum number of open connections to the database. If conn_max_idle is greater than 0 and the new conn_max_open is less than conn_max_idle, then conn_max_idle will be reduced to match the new conn_max_open limit. If value <= 0, then there is no limit on the number of open connections. The default is 0 (unlimited).

Optional manual configuration of AWS credentials to use. More information can be found in this document.

A database driver to use.

A Data Source Name to identify the target database.

Drivers

The following is a list of supported drivers, their placeholder style, and their respective DSN formats:

Please note that the postgres driver enforces SSL by default, you can override this with the parameter sslmode=disable if required.

The snowflake driver supports multiple DSN formats. Please consult the docs for more details. For key pair authentication, the DSN has the following format: <snowflake_user>@<snowflake_account>/<db_name>/<schema_name>?warehouse=<warehouse>&role=<role>&authenticator=snowflake_jwt&privateKey=<base64_url_encoded_private_key>, where the value for the privateKey parameter can be constructed from an unencrypted RSA private key file rsa_key.p8 using openssl enc -d -base64 -in rsa_key.p8 | basenc --base64url -w0 (you can use gbasenc insted of basenc on OSX if you install coreutils via Homebrew). If you have a password-encrypted private key, you can decrypt it using openssl pkcs8 -in rsa_key_encrypted.p8 -out rsa_key.p8. Also, make sure fields such as the username are URL-encoded.

The gocosmos driver is still experimental, but it has support for hierarchical partition keys as well as cross-partition queries. Please refer to the SQL notes for details.

Allows you to specify a custom endpoint for the AWS API. Default:

An optional field used to generate an IAM authentication token to connect to an Amazon Relational Database (RDS) DB instance. This will overwrite the Password in the DSN with the generated token only if the drivers are mysql or postgres. Default: false

An optional list of file paths containing SQL statements to execute immediately upon the first connection to the target database. This is a useful way to initialise tables before processing data. Glob patterns are supported, including super globs (double star).

Care should be taken to ensure that the statements are idempotent, and therefore would not cause issues when run multiple times after service restarts. If both init_statement and init_files are specified the init_statement is executed after the init_files.

If a statement fails for any reason a warning log will be emitted but the operation of this component will not be stopped.

An optional SQL statement to execute immediately upon the first connection to the target database. This is a useful way to initialise tables before processing data. Care should be taken to ensure that the statement is idempotent, and therefore would not cause issues when run multiple times after service restarts.

If both init_statement and init_files are specified the init_statement is executed after the init_files.

If the statement fails for any reason a warning log will be emitted but the operation of this component will not be stopped.

Whether to verify the database connection on startup by performing a simple ping, by default this is disabled. Default: false

The query to execute. The style of placeholder to use depends on the driver, some drivers require question marks (?) whereas others expect incrementing dollar signs ($1, $2, and so on) or colons (:1, :2 and so on). The style to use is outlined in this table:

The AWS region to target. Default:

An optional field that can be used to get the Username + Password from AWS Secrets Manager. This will overwrite the Username + Password in the DSN with the values from the Secret only if the driver is set to postgres.

An optional Bloblang mapping which should evaluate to an array of values matching in size to the number of placeholder arguments in the field where.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

Optional Fields that can be set to use Azure based authentication for Azure Postgres SQL

A list of columns to select.

An optional maximum number of connections in the idle connection pool. If conn_max_open is greater than 0 but less than the new conn_max_idle, then the new conn_max_idle will be reduced to match the conn_max_open limit. If value <= 0, no idle connections are retained. The default max idle connections is currently 2. This may change in a future release. Default: 2

An optional maximum amount of time a connection may be idle. Expired connections may be closed lazily before reuse. If value <= 0, connections are not closed due to a connections idle time.

An optional maximum amount of time a connection may be reused. Expired connections may be closed lazily before reuse. If value <= 0, connections are not closed due to a connections age.

An optional maximum number of open connections to the database. If conn_max_idle is greater than 0 and the new conn_max_open is less than conn_max_idle, then conn_max_idle will be reduced to match the new conn_max_open limit. If value <= 0, then there is no limit on the number of open connections. The default is 0 (unlimited).

Optional manual configuration of AWS credentials to use. More information can be found in this document.

A database driver to use.

A Data Source Name to identify the target database.

Drivers

The following is a list of supported drivers, their placeholder style, and their respective DSN formats:

Please note that the postgres driver enforces SSL by default, you can override this with the parameter sslmode=disable if required.

The snowflake driver supports multiple DSN formats. Please consult the docs for more details. For key pair authentication, the DSN has the following format: <snowflake_user>@<snowflake_account>/<db_name>/<schema_name>?warehouse=<warehouse>&role=<role>&authenticator=snowflake_jwt&privateKey=<base64_url_encoded_private_key>, where the value for the privateKey parameter can be constructed from an unencrypted RSA private key file rsa_key.p8 using openssl enc -d -base64 -in rsa_key.p8 | basenc --base64url -w0 (you can use gbasenc insted of basenc on OSX if you install coreutils via Homebrew). If you have a password-encrypted private key, you can decrypt it using openssl pkcs8 -in rsa_key_encrypted.p8 -out rsa_key.p8. Also, make sure fields such as the username are URL-encoded.

The gocosmos driver is still experimental, but it has support for hierarchical partition keys as well as cross-partition queries. Please refer to the SQL notes for details.

Allows you to specify a custom endpoint for the AWS API. Default:

An optional field used to generate an IAM authentication token to connect to an Amazon Relational Database (RDS) DB instance. This will overwrite the Password in the DSN with the generated token only if the drivers are mysql or postgres. Default: false

An optional list of file paths containing SQL statements to execute immediately upon the first connection to the target database. This is a useful way to initialise tables before processing data. Glob patterns are supported, including super globs (double star).

Care should be taken to ensure that the statements are idempotent, and therefore would not cause issues when run multiple times after service restarts. If both init_statement and init_files are specified the init_statement is executed after the init_files.

If a statement fails for any reason a warning log will be emitted but the operation of this component will not be stopped.

An optional SQL statement to execute immediately upon the first connection to the target database. This is a useful way to initialise tables before processing data. Care should be taken to ensure that the statement is idempotent, and therefore would not cause issues when run multiple times after service restarts.

If both init_statement and init_files are specified the init_statement is executed after the init_files.

If the statement fails for any reason a warning log will be emitted but the operation of this component will not be stopped.

Whether to verify the database connection on startup by performing a simple ping, by default this is disabled. Default: false

An optional prefix to prepend to the select query (before SELECT).

The AWS region to target. Default:

An optional field that can be used to get the Username + Password from AWS Secrets Manager. This will overwrite the Username + Password in the DSN with the values from the Secret only if the driver is set to postgres.

An optional suffix to append to the select query.

The table to select from.

An optional where clause to add. Placeholder arguments are populated with the args_mapping field. Placeholders should always be question marks, and will automatically be converted to dollar syntax when the postgres or clickhouse drivers are used.

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

The way in which the bytes of a data source should be converted into discrete messages, codecs are useful for specifying how large files or continuous streams of data might be processed in small chunks rather than loading it all in memory. It's possible to consume lines using a custom delimiter with the delim:x codec, where x is the character sequence custom delimiter. Codecs can be chained with /, for example a gzip compressed CSV file can be consumed with the codec gzip/csv.

Default: 1000000

The scanner by which the stream of bytes consumed will be broken out into individual messages. Scanners are useful for processing large sources of data without holding the entirety of it within memory. For example, the csv scanner allows you to process individual CSV rows without loading the entire CSV file in memory at once. Default: map[lines:map[]]

A list of arguments to provide the command. Default: []

The way in which messages should be consumed from the subprocess. Default: lines

The maximum expected size of an individual message. Default: 65536

The command to execute as a subprocess.

Whether the command should be re-executed each time the subprocess ends. Default: false

An API key for OAuth 2.0 authentication. It is recommended that you populate this field using environment variables.

An API secret for OAuth 2.0 authentication. It is recommended that you populate this field using environment variables.

A duration string indicating the maximum age of tweets to acquire when starting a search. Default: 5m

A cache resource to use for request pagination.

The key identifier used when storing the ID of the last tweet received. Default: last_tweet_id

The length of time (as a duration string) to wait between each search request. This field can be set empty, in which case requests are made at the limit set by the rate limit. This field also supports cron expressions. Default: 1m

A search expression to use.

An optional rate limit resource to restrict API requests with. Default:

An optional list of additional fields to obtain for each tweet, by default only the fields id and text are returned. For more info refer to the twitter API docs. Default: []

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. Default: true

Allows you to specify basic authentication.

Customise how websocket connection attempts are made.

A map of custom headers to add to the websocket handshake. Default: map[]

BETA: Allows you to specify JWT authentication.

Allows you to specify open authentication via OAuth version 1.

An optional message to send to the server upon connection.

An optional flag to indicate the data type of open_message. Default: binary