Components
Expanso provides 200+ components for building powerful data pipelines. Browse, search, and filter components to find exactly what you need.
Showing 184 of 184 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
Generates vector embeddings from text prompts, using the AWS Bedrock API.
aws_cloudwatch_logs
Input
Consumes log events from AWS CloudWatch Logs.
aws_dynamodb
Output
Inserts items into or deletes items from 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 and creates a batch of messages from each page of items.
azure_cosmosdb
Processor
Creates or updates messages as JSON documents in Azure CosmosDB.
azure_cosmosdb
Output
Creates or updates messages as JSON documents in Azure CosmosDB.
azure_data_lake_gen2
Output
Sends message parts as files to an Azure Data Lake Gen2 file system.
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 throughput statistics for processed messages, and provides a summary of those statistics over the lifetime of the processor.
bloblang
Processor
Executes a 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, execute a list of processors on the request messages, and, f...
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 for each message, allowing you to store or retrieve data within message payloads.
cache
Output
Stores each message in a cache.
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
Stops the current pipeline process with a fatal log message.
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. Drops sensor jitter, sub-significant drift, and flat-...
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 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 a text string, using the Vertex AI API.
generate
Input
Generates messages at a given interval using a Bloblang mapping executed without a context. This allows you to generate messages for testing your pipe...
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 into N batches, where each resulting batch contains a group of messages determined by a Bloblang query.
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 evalua...
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
input:
inproc
Output
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 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.
kafka_franz
Output
A Kafka output using the Franz Kafka client library.
log
Processor
Prints a log event for each message. Messages always remain unchanged. The log message can be set using function interpolations described here which a...
mapping
Processor
Executes a Bloblang mapping on messages, creating a new document that replaces (or filters) the original message.
metadata
Processor
Attaches Expanso runtime metadata — pipeline IDs, node info, runtime counters, custom fields — to messages, either as message metadata or spliced into...
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 mapping and directly transforms the contents of messages, mutating (or deleting) them.
mysql_cdc
Input
Streams changes from a MySQL database using its binary log to capture data updates.
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.
nats_stream
Output
Publish messages 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 and external tools.
ollama_embeddings
Processor
Generates vector embeddings from text, using the Ollama API.
ollama_moderation
Processor
Checks the responses of a large language model against safety and security standards, using the Ollama API.
openai_chat_completion
Processor
Generates responses to messages in a chat conversation, using the OpenAI API and external tools.
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 the OpenAI API.
openai_speech
Processor
Generates audio from a text description and other attributes, using the 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.
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`.
pg_stream
Input
Former name of the postgres_cdc input. Streams data changes from a PostgreSQL database using logical replication.
postgres_cdc
Input
Streams data 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 collection and filter the returned results.
qdrant
Output
Adds items to a Qdrant collection.
rate_limit
Processor
Throttles the throughput of a pipeline according to a specified `rate_limit` resource. Rate limits are shared across components and therefore apply gl...
read_until
Input
Reads messages from a child input until a consumed message passes a Bloblang query, at which point the input closes. It is also possible to configure ...
redis
Processor
Performs actions against Redis that aren't possible using a `cache` processor. Actions are
redis_hash
Output
Sets Redis hash objects using the HMSET 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.
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
Consumes topic data from one or more Kafka brokers.
redpanda
Output
Sends message data to Kafka brokers and waits for acknowledgement before propagating any acknowledgements back to the input.
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 behaviour 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 a schema registry.
schema_registry
Output
Publishes schemas to a schema registry.
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
Deterministically signs each pipeline message with the edge node's Ed25519 identity. Adds signature, keyId, and algorithm to message metadata, the JSO...
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
Allows Snowflake to ingest data from your data pipeline 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...
splunk_hec
Output
Writes messages to a Splunk HTTP Endpoint Collector.
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 custom SQL 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
Follows local files across rotation, emitting one message per line. Uses content fingerprinting for file identity and persists an ack-gated checkpoint...
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
Attempts a list of child processors and applies a separate list of catch processors if any of them fail.
unarchive
Processor
Unarchives messages according to the selected archive format into multiple messages within a batch.
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 against each batch of messages and executes child processors on them for as long as the query resolves to tru...
workflow
Processor
Executes a topology of [`branch` processors][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
- 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...
Quick Navigation
Use the categories above to browse components by type, or use the filters below to find components by:
- Cloud Provider: AWS, GCP, Azure, Generic
- Use Case: Streaming, Storage, Transform, Observability
- Difficulty: Beginner, Intermediate, Advanced
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
- ClickHouse for time-series data
- Snowflake and BigQuery for cloud data warehouses
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 200+ pre-built components across three categories: 61 inputs for receiving data, 74 processors for transforming data, and 71 outputs for sending data to destinations. 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.
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 filters above to search by name, category, or use case. You can also browse by cloud provider (AWS, GCP, Azure) or difficulty level. The most popular components section highlights commonly used components. If you're unsure, check our use cases section for real-world examples.