Components
Expanso provides 200+ components for building powerful data pipelines. Browse, search, and filter components to find exactly what you need.
Showing 185 of 185 components
archive
Processor
Archives all the messages of a batch into a single message according to the selected archive format.
awk
Processor
Executes an AWK program on messages. This processor is very powerful as it offers a range of custom functions for querying and mutating message conten...
aws_bedrock_chat
Processor
Generates responses to messages in a chat conversation, using the AWS Bedrock API.
aws_bedrock_embeddings
Processor
Computes vector embeddings on text, using the AWS Bedrock API.
aws_cloudwatch_logs
Input
Consumes log events from AWS CloudWatch Logs.
aws_dynamodb
Output
Inserts items into a DynamoDB table.
aws_dynamodb_cdc
Input
Reads change data capture (CDC) events from DynamoDB Streams.
aws_dynamodb_partiql
Processor
Executes a PartiQL expression against a DynamoDB table for each message.
aws_kinesis
Input
Receive messages from one or more Kinesis streams.
aws_kinesis
Output
Sends messages to a Kinesis stream.
aws_kinesis_firehose
Output
Sends messages to a Kinesis Firehose delivery stream.
aws_lambda
Processor
Invokes an AWS lambda for each message. The contents of the message is the payload of the request, and the result of the invocation will become the ne...
aws_s3
Input
Downloads objects within an Amazon S3 bucket, optionally filtered by a prefix, either by walking the items in the bucket or by streaming upload notifi...
aws_s3
Output
Sends message parts as objects to an Amazon S3 bucket. Each object is uploaded with the path specified with the `path` field.
aws_sns
Output
Sends messages to an AWS SNS topic.
aws_sqs
Input
Consume messages from an AWS SQS URL.
aws_sqs
Output
Sends messages to an SQS queue.
azure_blob_storage
Input
Downloads objects within an Azure Blob Storage container, optionally filtered by a prefix.
azure_blob_storage
Output
Sends message parts as objects to an Azure Blob Storage Account container. Each object is uploaded with the filename specified with the `container` fi...
azure_cosmosdb
Input
Executes a SQL query against [Azure CosmosDB](https://learn.microsoft.com/en-us/azure/cosmos-db/introduction) and creates a batch of messages from eac...
azure_cosmosdb
Processor
Creates or updates messages as JSON documents in [Azure CosmosDB](https://learn.microsoft.com/en-us/azure/cosmos-db/introduction).
azure_cosmosdb
Output
Creates or updates messages as JSON documents in [Azure CosmosDB](https://learn.microsoft.com/en-us/azure/cosmos-db/introduction).
azure_data_lake_gen2
Output
Sends message parts as files to an Azure Data Lake Gen2 filesystem. Each file is uploaded with the filename specified with the `path` field.
azure_queue_storage
Input
Dequeue objects from an Azure Storage Queue.
azure_queue_storage
Output
Sends messages to an Azure Storage Queue.
azure_table_storage
Input
Queries an Azure Storage Account Table, optionally with multiple filters.
azure_table_storage
Output
Stores messages in an Azure Table Storage table.
batched
Input
Consumes data from a child input and applies a batching policy to the stream.
benchmark
Processor
Logs basic throughput statistics of messages that pass through this processor.
bloblang
Processor
Executes a [Bloblang](/guides/bloblang) mapping on messages.
bounds_check
Processor
Removes messages (and batches) that do not fit within certain size boundaries.
branch
Processor
The `branch` processor allows you to create a new request message via a [Bloblang mapping](/guides/bloblang), execute a list of processors on the requ...
broker
Input
Allows you to combine multiple inputs into a single stream of data, where each input will be read in parallel.
broker
Output
Allows you to route messages to multiple child outputs using a range of brokering patterns.
cache
Processor
Performs operations against a [cache resource](/components/caches/) for each message, allowing you to store or retrieve data within message payloads.
cache
Output
Stores each message in a [cache](/components/caches/).
cached
Processor
Cache the result of applying one or more processors to messages identified by a key. If the key already exists within the cache the contents of the me...
catch
Processor
Applies a list of child processors _only_ when a previous processing step has failed.
command
Processor
Executes a command for each message.
compress
Processor
Compresses messages according to the selected algorithm. Supported compression algorithms are: [flate gzip lz4 pgzip snappy zlib]
crash
Processor
Crashes the process using a fatal log message. The log message can be set using function interpolations described in [Bloblang queries](/guides/pipeli...
csv
Input
Reads one or more CSV files as structured records following the format described in RFC 4180.
deadband
Processor
Forwards a numeric reading only when it has changed meaningfully since the last forwarded value.
decompress
Processor
Decompresses messages according to the selected algorithm. Supported decompression algorithms are: [bzip2 flate gzip lz4 pgzip snappy zlib]
dedupe
Processor
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.
drop
Output
Drops all messages.
drop_on
Output
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 o...
dynamic
Input
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 interfac...
dynamic
Output
A special broker type where the outputs are identified by unique labels and can be created, changed and removed during runtime via a REST API.
fallback
Output
Attempts to send each message to a child output, starting from the first output on the list. If an output attempt fails then the next output in the li...
file
Input
Consumes data from files on disk, emitting messages according to a chosen codec.
file
Output
Writes messages to files on disk based on a chosen codec.
for_each
Processor
A processor that applies a list of child processors to messages of a batch as though they were each a batch of one message.
gcp_bigquery
Output
Sends messages as new rows to a Google Cloud BigQuery table.
gcp_bigquery_select
Input
Executes a `SELECT` query against BigQuery and creates a message for each row received.
gcp_bigquery_select
Processor
Executes a `SELECT` query against BigQuery and replaces messages with the rows returned.
gcp_cloud_storage
Input
Downloads objects within a Google Cloud Storage bucket, optionally filtered by a prefix.
gcp_cloud_storage
Output
Sends message parts as objects to a Google Cloud Storage bucket. Each object is uploaded with the path specified with the `path` field.
gcp_pubsub
Input
Consumes messages from a GCP Cloud Pub/Sub subscription.
gcp_pubsub
Output
Sends messages to a GCP Cloud Pub/Sub topic. [Metadata](/guides/pipelines/metadata) from messages are sent as attributes.
gcp_vertex_ai_chat
Processor
Generates responses to messages in a chat conversation, using the Vertex AI API.
gcp_vertex_ai_embeddings
Processor
Generates vector embeddings to represent input text, using the Vertex AI API.
generate
Input
Generates messages at a given interval using a [Bloblang](/guides/bloblang) mapping executed without a context. This allows you to generate messages f...
grok
Processor
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 repla...
group_by
Processor
Splits a [batch of messages](/guides/pipelines/batching) into N batches, where each resulting batch contains a group of messages determined by a [Blob...
group_by_value
Processor
Splits a batch of messages into N batches, where each resulting batch contains a group of messages determined by a [function interpolated string](/gui...
http
Processor
Performs an HTTP request using a message batch as the request body, and replaces the original message parts with the body of the response.
http_client
Input
Connects to a server and continuously performs requests for a single message.
http_client
Output
Sends messages to an HTTP server.
http_server
Input
Receive messages POSTed over HTTP(S). HTTP 2.0 is supported when using TLS, which is enabled when key and cert files are specified.
http_server
Output
Sets up an HTTP server that will send messages over HTTP(S) GET requests. HTTP 2.0 is supported when using TLS, which is enabled when key and cert fil...
inproc
Input
inproc
Output
insert_part
Processor
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.
jmespath
Processor
Executes a [JMESPath query](http://jmespath.org/) on JSON documents and replaces the message with the resulting document.
jq
Processor
Transforms and filters messages using jq queries.
json_schema
Processor
Checks messages against a provided JSONSchema definition but does not change the payload under any circumstances. If a message does not match the sche...
kafka
Input
Connects to Kafka brokers and consumes one or more topics.
kafka
Output
The kafka output type writes a batch of messages to Kafka brokers and waits for acknowledgement before propagating it back to the input.
kafka_franz
Input
A Kafka input using the [Franz Kafka client library](https://github.com/twmb/franz-go).
kafka_franz
Output
A Kafka output using the [Franz Kafka client library](https://github.com/twmb/franz-go).
log
Processor
Prints a log event for each message. Messages always remain unchanged. The log message can be set using function interpolations described in [Bloblang...
mapping
Processor
Executes a [Bloblang](/guides/bloblang) mapping on messages, creating a new document that replaces (or filters) the original message.
metadata
Processor
Attaches Expanso runtime metadata to messages.
metric
Processor
Emit custom metrics by extracting values from messages.
mqtt
Input
Subscribe to topics on MQTT brokers.
mqtt
Output
Pushes messages to an MQTT broker.
msgpack
Processor
Converts messages to or from the [MessagePack](https://msgpack.org/) format.
mutation
Processor
Executes a [Bloblang](/guides/bloblang) mapping and directly transforms the contents of messages, mutating (or deleting) them.
mysql_cdc
Input
Enables MySQL streaming for Expanso Edge.
nats
Input
Subscribe to a NATS subject.
nats
Output
Publish to an NATS subject.
nats_jetstream
Input
Reads messages from NATS JetStream subjects.
nats_jetstream
Output
Write messages to a NATS JetStream subject.
nats_kv
Input
Watches for updates in a NATS key-value bucket.
nats_kv
Processor
Perform operations on a NATS key-value bucket.
nats_kv
Output
Put messages in a NATS key-value bucket.
nats_request_reply
Processor
Sends a message to a NATS subject and expects a reply, from a NATS subscriber acting as a responder, back.
nats_stream
Input
Subscribe to a NATS Stream subject. Joining a queue is optional and allows multiple clients of a subject to consume using queue semantics.
nats_stream
Output
Publish to a NATS Stream subject.
noop
Processor
Noop is a processor that does nothing, the message passes through unchanged. Why? Sometimes doing nothing is the braver option.
ollama_chat
Processor
Generates responses to messages in a chat conversation, using the Ollama API.
ollama_embeddings
Processor
Generates vector embeddings from text, using the Ollama API.
ollama_moderation
Processor
Generates responses to messages in a chat conversation, using the Ollama API.
opcua
Input
OPC UA input plugin
opcua
Output
OPC UA output plugin
openai_chat_completion
Processor
Generates responses to messages in a chat conversation, using the OpenAI API.
openai_embeddings
Processor
Generates vector embeddings to represent input text, using the OpenAI API.
openai_image_generation
Processor
Generates an image from a text description and other attributes, using OpenAI API.
openai_speech
Processor
Generates audio from a text description and other attributes, using OpenAI API.
openai_transcription
Processor
Generates a transcription of spoken audio in the input language, using the OpenAI API.
openai_translation
Processor
Translates spoken audio into English, using the OpenAI API.
opensearch
Output
Publishes messages into an Elasticsearch index. If the index does not exist then it is created with a dynamic mapping.
parallel
Processor
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`...
parquet
Input
Reads and decodes [Parquet files](https://parquet.apache.org/docs/) into a stream of structured messages.
parquet
Processor
Converts batches of documents to or from [Parquet files](https://parquet.apache.org/docs/).
parquet_decode
Processor
Decodes [Parquet files](https://parquet.apache.org/docs/) into a batch of structured messages.
parquet_encode
Processor
Encodes [Parquet files](https://parquet.apache.org/docs/) from a batch of structured messages.
parse_log
Processor
Parses common log formats into structured data. This is easier and often much faster than [`grok`](/components/processors/grok).
pg_stream
Input
Streams changes from a PostgreSQL database using logical replication.
postgres_cdc
Input
Streams changes from a PostgreSQL database using logical replication.
processors
Processor
A processor grouping several sub-processors.
protobuf
Processor
Performs conversions to or from a protobuf message. This processor uses reflection, meaning conversions can be made directly from the target .proto fi...
qdrant
Processor
Query items within a [Qdrant](https://qdrant.tech/) collection.
qdrant
Output
Adds items to a [Qdrant](https://qdrant.tech/) collection
rate_limit
Processor
Throttles the throughput of a pipeline according to a specified [`rate_limit`](/components/rate_limits/) resource. Rate limits are shared across compo...
read_until
Input
Reads messages from a child input until a consumed message passes a [Bloblang query](/guides/bloblang), at which point the input closes. It is also po...
redis
Processor
Performs actions against Redis that aren't possible using a [`cache`](/components/processors/cache) processor. Actions are performed for each message ...
redis_hash
Output
Sets Redis hash objects using the HSET command.
redis_list
Input
Pops messages from the beginning of a Redis list using the BLPop command.
redis_list
Output
Pushes messages onto the end of a Redis list (which is created if it doesn't already exist) using the RPUSH command.
redis_pubsub
Input
Consume from a Redis publish/subscribe channel using either the SUBSCRIBE or PSUBSCRIBE commands.
redis_pubsub
Output
Publishes messages through the Redis PubSub model. It is not possible to guarantee that messages have been received.
redis_scan
Input
Scans the set of keys in the current selected database and gets their values, using the Scan and Get commands.
redis_script
Processor
Performs actions against Redis using [LUA scripts](https://redis.io/docs/manual/programmability/eval-intro/).
redis_streams
Input
Pulls messages from Redis (v5.0+) streams with the XREADGROUP command. The `client_id` should be unique for each consumer of a group.
redis_streams
Output
Pushes messages to a Redis (v5.0+) Stream (which is created if it doesn't already exist) using the XADD command.
redpanda
Input
A Kafka input using the [Franz Kafka client library](https://github.com/twmb/franz-go).
redpanda
Output
A Kafka output using the [Franz Kafka client library](https://github.com/twmb/franz-go).
reject
Output
Rejects all messages, treating them as though the output destination failed to publish them.
reject_errored
Output
Rejects messages that have failed their processing steps, resulting in nack behavior at the input level, otherwise sends them to a child output.
resource
Input
Resource is an input type that channels messages from a resource input, identified by its name.
resource
Processor
Resource is a processor type that runs a processor resource identified by its label.
resource
Output
Resource is an output type that channels messages to a resource output, identified by its name.
retry
Processor
Attempts to execute a series of child processors until success.
retry
Output
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 m...
schema_registry
Input
Reads schemas from SchemaRegistry.
schema_registry
Output
Publishes schemas to SchemaRegistry.
select_parts
Processor
Cherry pick a set of messages from a batch by their index. Indexes larger than the number of messages are simply ignored.
sequence
Input
Reads messages from a sequence of child inputs, starting with the first and once that input gracefully terminates starts consuming from the next, and ...
signature
Processor
Signs each message's payload with the edge node's Ed25519 identity keypair.
sleep
Processor
Sleep for a period of time specified as a duration string for each message. This processor will interpolate functions within the `duration` field, you...
snowflake_put
Output
Sends messages to Snowflake stages and, optionally, calls Snowpipe to load this data into one or more tables.
snowflake_streaming
Output
Ingest data into Snowflake using Snowpipe Streaming.
socket
Input
Connects to a tcp or unix socket and consumes a continuous stream of messages.
socket
Output
Connects to a (tcp/udp/unix) server and sends a continuous stream of data, dividing messages according to the specified codec.
socket_server
Input
Creates a server that receives a stream of messages over a TCP, UDP or Unix socket.
split
Processor
Breaks message batches (synonymous with multiple part messages) into smaller batches. The size of the resulting batches are determined either by a dis...
sql
Processor
Runs an arbitrary SQL query against a database and (optionally) returns the result as an array of objects, one for each row returned.
sql
Output
Executes an arbitrary SQL query for each message.
sql_insert
Processor
Inserts rows into an SQL database for each message, and leaves the message unchanged.
sql_insert
Output
Inserts a row into an SQL database for each message.
sql_raw
Input
Executes a select query and creates a message for each row received.
sql_raw
Processor
Runs an arbitrary SQL query against a database and (optionally) returns the result as an array of objects, one for each row returned.
sql_raw
Output
Executes an arbitrary SQL query for each message.
sql_select
Input
Executes a select query and creates a message for each row received.
sql_select
Processor
Runs an SQL select query against a database and returns the result as an array of objects, one for each row returned, containing a key for each column...
stdin
Input
Consumes data piped to stdin, chopping it into individual messages according to the specified scanner.
stdout
Output
Prints messages to stdout as a continuous stream of data.
subprocess
Input
Executes a command, runs it as a subprocess, and consumes messages from it over stdout.
subprocess
Processor
Executes a command as a subprocess and, for each message, will pipe its contents to the stdin stream of the process followed by a newline.
subprocess
Output
Executes a command, runs it as a subprocess, and writes messages to it over stdin.
switch
Processor
Conditionally processes messages based on their contents.
switch
Output
The switch output type allows you to route messages to different outputs based on their contents.
sync_response
Processor
Adds the payload in its current state as a synchronous response to the input source, where it is dealt with according to that specific input type.
sync_response
Output
Returns the final message payload back to the input origin of the message, where it is dealt with according to that specific input type.
tail
Input
Tails local files with rotation handling and persistent checkpointing.
try
Processor
Executes a list of child processors on messages only if no prior processors have failed (or the errors have been cleared).
try_catch
Processor
Executes a list of child `processors` on each message and, if any of them fail, executes a separate list of `catch` processors to recover from or reac...
unarchive
Processor
Unarchives messages according to the selected archive format into multiple messages within a [batch](/guides/pipelines/batching).
websocket
Input
Connects to a websocket server and continuously receives messages.
websocket
Output
Sends messages to an HTTP server via a websocket connection.
while
Processor
A processor that checks a [Bloblang query](/guides/bloblang) against each batch of messages and executes child processors on them for as long as the q...
workflow
Processor
Executes a topology of [`branch` processors](/components/processors/branch), performing them in parallel where possible.
xml
Processor
Parses messages as an XML document, performs a mutation on the data, and then overwrites the previous contents with the new value.
Component Categories
📥 Inputs
Receive data from various sources including:
- Messaging Systems: Kafka, NATS, RabbitMQ, Redis
- Cloud Storage: AWS S3, GCS, Azure Blob
- Databases: PostgreSQL, MySQL, MongoDB (nightly requirement)
- HTTP: Webhooks, REST APIs
- Files: Local files, SFTP
- And many more...
⚙️ Processors
Transform and manipulate your data:
- Mapping: Transform data with Bloblang
- Filtering: Route and filter messages
- Aggregation: Window, batch, and aggregate
- Enrichment: Lookup and enrich data
- Encoding: JSON, CSV, Avro, Protobuf
- And many more...
📤 Outputs
Send data to destinations:
- Messaging Systems: Kafka, NATS, RabbitMQ, Redis
- Cloud Storage: AWS S3, GCS, Azure Blob
- Databases: PostgreSQL, MySQL, Elasticsearch
- HTTP: Webhooks, REST APIs
- Observability: Prometheus, Datadog, New Relic
- And many more...
🧩 Supporting Components
Round out pipelines with:
- Caches: Key/value stores for deduplication and enrichment
- Buffers: Decouple ingestion from processing
- Rate Limits: Throttle how fast components fire
- Scanners: Split raw input streams into messages
Quick Navigation
Use the catalog above to search by name or description, and to filter by:
- Type: Inputs, Processors, Outputs
- Category: Services, Utility, Parsing, AI, Integration, Composition, Network, Mapping, Local, and the cloud providers AWS, Azure, and GCP
- Status: Stable, Beta, Deprecated
Most Popular Components
Here are some of the most commonly used components:
- Kafka Input/Output: Stream data to/from Apache Kafka
- Bloblang Processor: Transform data with Bloblang mapping language
- Database Components: Connect to MySQL, PostgreSQL, SQLite and more
- S3 Output: Store data in AWS S3 buckets
- HTTP Server Input: Receive data via HTTP webhooks
Database Integration
Connect to databases for edge analytics, caching, and data synchronization:
- Database Connectivity Guide: Complete guide to MySQL, PostgreSQL, SQLite
- SQL Select Input: Read data from database tables
- SQL Insert Output: Write data to database tables
- SQL Raw Input/Output: Execute custom SQL queries
- Database Patterns: Real-world integration examples
Popular database integrations:
- MySQL and PostgreSQL for cloud databases
- SQLite for local caching, offline storage, and edge analytics
- MongoDB for document storage (nightly requirement)
- ClickHouse for time-series data
- Snowflake and BigQuery for cloud data warehouses
MongoDB availability
As of September 10, 2026, MongoDB input, output, processor, and cache components are available in nightly v2.1.22-nightly.134.g0858e17a. They are not included in stable v2.1.21, which the default installer selects. MongoDB pipelines require an edge-node release that includes these components.
Getting Started
New to Expanso? Start with these guides:
- Quick Start: Build your first pipeline in 5 minutes
- Core Concepts: Understand how components work together
- Bloblang Guide: Learn the transformation language
Frequently Asked Questions
How many components are available in Expanso?
Expanso provides more than 200 pre-built components across seven categories: inputs for receiving data, processors for transforming it, outputs for sending it to destinations, plus caches, buffers, rate limits, and scanners that support them. All components are production-ready and maintained by the Expanso team.
Can I use custom components or extend existing ones?
While you can't add completely custom components, you can achieve most custom logic using the Mapping processor with Bloblang language. Bloblang provides extensive functions for parsing, transforming, and routing data. For more complex needs, you can chain multiple components together or contact support for enterprise custom component development.
What's the difference between inputs, processors, and outputs?
Inputs receive data from sources (like Kafka, S3, or HTTP). Processors transform data in transit (filtering, mapping, enrichment). Outputs send data to destinations (databases, cloud storage, APIs). Every pipeline needs at least one input and one output, with zero or more processors in between. Pipelines can also use supporting components: caches, buffers, rate limits, and scanners.
Are all components available in the free tier?
Yes! All 200+ components are available in every tier, including the free tier. There are no feature restrictions on which components you can use - only limits on data volume and number of agents. Check our pricing page for tier details.
How do I find the right component for my use case?
Use the component catalog above to search by name or description, and to filter by category, type, and status. Cloud providers are categories too, so AWS, GCP, and Azure components can be filtered as a group. The most popular components section highlights commonly used components. If you're unsure, check our use cases section for real-world examples.