Components Catalog

Transform and process message data

Connects to an AMQP (0.91) queue. AMQP is a messaging protocol used by various message brokers, including RabbitMQ.

Reads messages from an AMQP (1.0) server.

Archives all the messages of a batch into a single message according to the selected archive format.

Performs Avro based operations on messages based on a schema.

Executes an AWK program on messages.

Generates responses to messages in a chat conversation, using the AWS Bedrock API.

Generates vector embeddings from text prompts, using the AWS Bedrock API.

Send metrics to AWS CloudWatch using the PutMetricData endpoint.

Stores key/value pairs as a single document in a DynamoDB table.

Receive data from external sources

Executes a PartiQL expression against a DynamoDB table for each message.

Receive messages from one or more Kinesis streams.

Sends messages to a Kinesis Firehose delivery stream.

Invokes an AWS lambda for each message.

Stores each item in an S3 bucket as a file, where an item ID is the path of the item within the bucket.

Sends messages to an AWS SNS topic.

Consume messages from an AWS SQS URL.

Downloads objects within an Azure Blob Storage container, optionally filtered by a prefix.

Executes a SQL query against Azure CosmosDB and creates a batch of messages from each page of items.

Sends message parts as files to an Azure Data Lake Gen2 file system. Each file is uploaded with the file name specified in the path field.

Dequeue objects from an Azure Storage Queue.

Queries an Azure Storage Account Table, optionally with multiple filters.

Consumes data from a child input and applies a batching policy to the stream.

Reads messages from a Beanstalkd queue.

Logs throughput statistics for processed messages, and provides a summary of those statistics over the lifetime of the processor.

Executes a Bloblang mapping on messages.

Removes messages (and batches) that do not fit within certain size boundaries.

The branch processor allows you to create a new request message via a Bloblang mapping, execute a list of processors on the request messages, and, finally, map the result back into the source message using another mapping.

Allows you to combine multiple inputs into a single stream of data, where each input will be read in parallel.

Stores each message in a cache.

Cache the result of applying one or more processors to messages identified by a key.

Executes a find query and creates a message for each row received.

Applies a list of child processors only when a previous processing step has failed.

Split an input stream into chunks of a given number of bytes.

Listens to a CockroachDB Core Changefeed and creates a message for each row received. Each message is a json object looking like:

Generates responses to messages in a chat conversation, using the Cohere API and external tools.

Generates vector embeddings to represent input text, using the Cohere API.

Sends document strings to the Cohere API, which returns them ranked by their relevance to a specified query.

Executes a command for each message.

Compresses messages according to the selected algorithm. Supported compression algorithms are: [flate gzip lz4 pgzip snappy zlib]

Use a Couchbase instance as a cache.

Transform and process message data

Reads one or more CSV files as structured records following the format described in RFC 4180.

Inserts items into a CyborgDB encrypted vector index.

Writes a batch of messages to any graph database that supports the Neo4j and Bolt URI schemes.

Decompresses messages according to the selected algorithm. Supported decompression algorithms are: [bzip2 flate gzip lz4 pgzip snappy zlib]

Deduplicates messages by storing a key value in a cache using the add operator. If the key already exists within the cache it is dropped.

Consumes messages posted in a Discord channel.

Send data to external destinations

Attempts to write messages to a child output and if the write fails for one of a list of configurable reasons the message is dropped (acked) instead of being reattempted (or nacked).

A special broker type where the inputs are identified by unique labels and can be created, changed and removed during runtime via a REST HTTP interface.

Publishes messages into an Elasticsearch index. If the index does not exist, this output creates it using dynamic mapping.

Publishes messages into an Elasticsearch index. If the index does not exist then it is created with a dynamic mapping.

Attempts to send each message to a child output, starting from the first output on the list.

Invoke a function within a shared library as a processing step.

Stores each item in a directory as a file, where an item ID is the path relative to the configured directory.

A processor that applies a list of child processors to messages of a batch as though they were each a batch of one message.

Receive data from external sources

Inserts message data as new rows in a Google Cloud BigQuery table.

Executes a SELECT query against BigQuery and creates a message for each row received.

Use a Google Cloud Storage bucket as a cache.

Send tracing events to a Google Cloud Trace.

Consumes messages from a GCP Cloud Pub/Sub subscription.

Creates an input that consumes from a spanner change stream.

Generates responses to messages in a chat conversation, using the Vertex API AI.

Generates vector embeddings to represent a text string, using the Vertex AI API.

Generates messages at a given interval using a Bloblang mapping executed without a context.

Clones a Git repository, reads its contents, then polls for new commits at a configurable interval. Any updates are emitted as new messages.

Downloads files from Google Drive that contain matching file IDs.

Lists labels for files on a Google Drive.

Searches Google Drive for files that match a specified query and emits the results as a batch of messages.

Parses messages into a structured format by attempting to apply a list of Grok expressions, the first expression to result in at least one value replaces the original message with a JSON object containing the values.

Splits a batch of messages into N batches, where each resulting batch contains a group of messages determined by a Bloblang query.

Splits a batch of messages into N batches, where each resulting batch contains a group of messages determined by a function interpolated string evaluated per message.

Reads files from a HDFS directory, where each discrete file will be consumed as a single message payload.

Performs a HTTP request using a message batch as the request body, and replaces the original message parts with the body of the response.

Connects to a server and continuously requests single messages.

Receive messages sent over HTTP using POST requests. HTTP 2.0 is supported when using TLS, which is enabled when key and cert files are specified.

Send metrics to InfluxDB 1.x using the /write endpoint.

Directly connect to an output within a Redpanda Connect process by referencing it by a chosen ID.

Insert a new message into a batch at an index. If the specified index is greater than the length of the existing batch it will be appended to the end.

Send tracing events to a Jaeger agent or collector.

Executes a JavaScript code block or file for each message.

Queries Jira resources and returns structured data.

Executes a JMESPath query on JSON documents and replaces the message with the resulting document.

Transforms and filters messages using jq queries.

Serves metrics as JSON object with the service wide HTTP service at the endpoints /stats and /metrics.

Consumes a stream of one or more JSON elements within a top level array.

Consumes a stream of one or more JSON documents.

Checks messages against a provided JSONSchema definition but does not change the payload under any circumstances.

Connects to Kafka brokers and consumes one or more topics.

A Kafka input using the Franz Kafka client library.

Use this connector in conjunction with the legacyredpandamigrator output to migrate topics between Apache Kafka brokers.

Reads consumer group offsets for a specified set of topics using the Franz Kafka client library.

Split an input stream into a message per line of data.

The local rate limit is a simple X every Y type rate limit that can be shared across any number of components within the pipeline but does not support distributed rate limits across multiple running instances of Benthos.

Prints a log event for each message.

Prints aggregated metrics through the logger.

Stores key/value pairs in a lru in-memory cache. This cache is therefore reset every time the service restarts.

Executes a Bloblang mapping on messages, creating a new document that replaces (or filters) the original message.

Connects to a cluster of memcached services, a prefix can be specified to allow multiple cache types to share a memcached cluster under different namespaces.

Stores consumed messages in memory and acknowledges them at the input level.

Emit custom metrics by extracting values from messages.

Enables Change Data Capture by consuming from Microsoft SQL Server's change tables.

Use a MongoDB instance as a cache.

Streams data changes from a MongoDB replica set, using MongoDB's change streams to capture data updates.

Subscribe to topics on MQTT brokers.

Converts messages to or from the MessagePack format.

Combines multiple caches as levels, performing read-through and write-through operations across them.

Executes a Bloblang mapping and directly transforms the contents of messages, mutating (or deleting) them.

Streams data changes from a MySQL database, using MySQL's binary log to capture data updates.

Consumes messages via Nanomsg sockets (scalability protocols).

Subscribe to a NATS subject.

Reads messages from NATS JetStream subjects.

Cache key/value pairs in a NATS key-value bucket.

Sends a message to a NATS subject and expects a reply back from a NATS subscriber acting as a responder.

Subscribe to a NATS Stream subject. Joining a queue is optional and allows multiple clients of a subject to consume using queue semantics.

Do not buffer messages. This is the default and most resilient configuration.

Noop is a cache that stores nothing, all gets returns not found. Why? Sometimes doing nothing is the braver option.

Subscribe to an NSQ instance topic and channel.

Uses Ockam to decrypt and read end-to-end encrypted messages from Kafka topics.

Generates responses to messages in a chat conversation using the Ollama API and external tools.

Generates vector embeddings from text, using the Ollama API.

Generates responses to messages in a chat conversation using the Ollama API, and checks the responses to make sure they do not violate safety or security standards.

Send tracing events to an Open Telemetry collector.

Generates responses to messages in a chat conversation, using the OpenAI API and external tools.

Generates vector embeddings to represent input text, using the OpenAI API.

Generates an image from a text description and other attributes, using OpenAI API.

Generates audio from a text description and other attributes, using OpenAI API.

Generates a transcription of spoken audio in the input language, using the OpenAI API.

Translates spoken audio into English, using the OpenAI API.

Publishes messages into an Elasticsearch index. If the index does not exist then it is created with a dynamic mapping.

Receive OpenTelemetry traces, logs, and metrics via OTLP/gRPC protocol.

Receive OpenTelemetry traces, logs, and metrics via OTLP/HTTP protocol.

A processor that applies a list of child processors to messages of a batch as though they were each a batch of one message (similar to the for_each processor), but where each message is processed in parallel.

Reads and decodes Parquet files into a stream of structured messages.

Decodes Parquet files into a batch of structured messages.

Encodes Parquet files from a batch of structured messages.

Parses common log <> into <>.

Receive data from external sources

Inserts items into a Pinecone index.

Streams data changes from a PostgreSQL database using logical replication.

A processor grouping several sub-processors.

Host endpoints (/metrics and /stats) for Prometheus scraping.

Handles conversions between JSON documents and protobuf messages using reflection, which allows you to make conversions from or to the target .

Reads messages from an Apache Pulsar server.

Output for publishing messages to Pusher API (https://pusher.com)

Adds items to a Qdrant collection

Pushes messages to a QuestDB table.

Throttles the throughput of a pipeline according to a specified rate_limit resource.

Split an input stream into segments matching against a regular expression.

Reads messages from a child input until a consumed message passes a Bloblang query, at which point the input closes.

Use a Redis instance as a cache. The expiration can be set to zero or an empty string in order to set no expiration.

Sets Redis hash objects using the HMSET command.

Pops messages from the beginning of a Redis list using the BLPop command.

Consume from a Redis publish/subscribe channel using either the SUBSCRIBE or PSUBSCRIBE commands.

Scans the set of keys in the current selected database and gets their values, using the Scan and Get commands.

Performs actions against Redis using LUA scripts.

Pulls messages from Redis (v5.0+) streams with the XREADGROUP command. The client_id should be unique for each consumer of a group.

A Kafka cache using the https://github.com/twmb/franz-go[Franz Kafka client library^].

Consumes data from a Redpanda (Kafka) broker, using credentials from a common redpanda configuration block.

Executes a Redpanda Data Transform as a processor

Unified Kafka consumer for migrating data between Kafka/Redpanda clusters.

The redpandamigratorbundle input reads messages and schemas from an Apache Kafka or Redpanda cluster.

Rejects all messages, treating them as though the output destination failed to publish them.

Rejects messages that have failed their processing steps, resulting in nack behavior at the input level, otherwise sends them to a child output.

Resource is an input type that channels messages from a resource input, identified by its name.

Attempts to write messages to a child output and if the write fails for any reason the message is retried either until success or, if the retries or max elapsed time fields are non-zero, either is reached.

Stores key/value pairs in a map held in the memory-bound Ristretto cache.

Reads schemas from a schema registry.

Automatically decodes and validates messages with schemas from a Confluent Schema Registry service.

Automatically encodes and validates messages with schemas from a Confluent Schema Registry service.

Cherry pick a set of messages from a batch by their index. Indexes larger than the number of messages are simply ignored.

Captures log events from messages and submits them to Sentry.

Reads messages from a sequence of child inputs, starting with the first and once that input gracefully terminates starts consuming from the next, and so on.

Consumes files from an SFTP server.

Skip one or more byte order marks for each opened child scanner.

Connects to Slack using Socket Mode, and can receive events, interactions (automated and user-initiated), and slash commands.

Posts a new message to a Slack channel using the Slack API method chat.postMessage.

Add or remove an emoji reaction to a Slack message.

Reads a Slack thread using the Slack API method conversations.replies.

Returns the full profile of all users in your Slack organization using the API method users.

Sleep for a period of time specified as a duration string for each message.

Sends messages to Snowflake stages and, optionally, calls Snowpipe to load this data into one or more tables.

Allows Snowflake to ingest data from your data pipeline using Snowpipe Streaming.

Connects to a tcp or unix socket and consumes a continuous stream of messages.

Creates a server that receives a stream of messages over a TCP, UDP or Unix socket.

Consumes messages from the Watch API of a SpiceDB instance.

Breaks message batches (synonymous with multiple part messages) into smaller batches.

Consumes messages from Splunk.

Publishes messages to a Splunk HTTP Endpoint Collector (HEC).

Uses an SQL database table as a destination for storing cache key/value items.

Execute SQL queries across multiple database engines

Execute SQL queries across multiple database engines

Execute SQL queries across multiple database engines

Execute SQL queries across multiple database engines

Execute SQL queries across multiple database engines

Execute SQL queries across multiple database engines

Execute SQL queries across multiple database engines

Execute SQL queries across multiple database engines

Execute SQL queries across multiple database engines

Inserts a row into an SQL database for each message.

Executes a select query and creates a message for each row received.

Executes a select query and creates a message for each row received.

Stores messages in an SQLite database and acknowledges them at the input level.

Pushes metrics using the StatsD protocol. Supported tagging formats are 'none', 'datadog' and 'influxdb'.

Consumes data piped to stdin, chopping it into individual messages according to the specified scanner.

Prints messages to stdout as a continuous stream of data.

Executes a command, runs it as a subprocess, and consumes messages from it over stdout.

The switch output type allows you to route messages to different outputs based on their contents.

Returns the final message payload back to the input origin of the message, where it is dealt with according to that specific input type.

Chops a stream of messages into tumbling or sliding windows of fixed temporal size, following the system clock.

Consume a tar archive file by file.

Breaks down text-based message content into manageable chunks using a configurable strategy.

Introduced in version 4.65.0

Executes a streaming or table query on Timeplus Enterprise (Cloud or Self-Hosted) or the timeplusd component, and creates a structured message for each table row received.

Read the input stream all the way until the end and deliver it as a single message.

Executes a list of child processors on messages only if no prior processors have failed (or the errors have been cleared).

Stores key/value pairs in a ttlru in-memory cache. This cache is therefore reset every time the service restarts.

Consumes tweets matching a given search using the Twitter recent search V2 API.

Unarchives messages according to the selected archive format into multiple messages within a batch.

Executes a function exported by a WASM module for each message.

Connects to a websocket server and continuously receives messages.

A processor that checks a Bloblang query against each batch of messages and executes child processors on them for as long as the query resolves to true.

Executes a topology of branch processors, performing them in parallel where possible.

Parses messages as an XML document, performs a mutation on the data, and then overwrites the previous contents with the new value.

Consumes messages from a ZeroMQ socket.