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.

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.

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

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

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.

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.

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.

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.

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

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

Inserts items into a CyborgDB encrypted vector index.

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.

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).

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

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

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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

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.

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.

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.

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

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.