Skip to main content

Processors

A processor is a unit of logic that transforms, filters, or otherwise processes messages as they flow through your pipeline:

pipeline:
  processors:
    - mapping: |
        root.message = this.content.uppercase()
        root.meta.processed_at = now()

    - filter:
        check: 'this.type == "important"'

Processors can be configured at the input, pipeline, or output level. They are applied in the order they are defined.

Processor Pipelines

Processors can be nested to create complex transformation logic:

pipeline:
  processors:
    - branch:
        processors:
          - cache:
              resource: user_cache
              operator: get
              key: ${! json("user_id") }
        result_map: 'root.user = this'

    - mapping: |
        root.enriched = this.merge(this.user)

Error Handling

Processors can fail during execution. Use the catch processor to handle errors gracefully:

pipeline:
  processors:
    - try:
        - mapping: 'root = this.parse_json()'
    - catch:
        - log:
            message: "Failed to parse: ${! content() }"

Browse Processors

Category:

Status:

Showing 80 of 185 components

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

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

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

aws_bedrock_embeddings

Computes vector embeddings on text, using the AWS Bedrock API.

aws_dynamodb_partiql

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

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

Creates or updates messages as JSON documents in [Azure CosmosDB](https://learn.microsoft.com/en-us/azure/cosmos-db/introduction).

Logs basic throughput statistics of messages that pass through this processor.

Executes a [Bloblang](/guides/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](/guides/bloblang), execute a list of processors on the requ...

Performs operations against a [cache resource](/components/caches/) for each message, allowing you to store or retrieve data within message payloads.

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

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

Executes a command for each message.

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

Crashes the process using a fatal log message. The log message can be set using function interpolations described in [Bloblang queries](/guides/pipeli...

Forwards a numeric reading only when it has changed meaningfully since the last forwarded value.

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.

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_select

Executes a `SELECT` query against BigQuery and replaces messages with the rows returned.

gcp_vertex_ai_chat

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

gcp_vertex_ai_embeddings

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

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

Splits a [batch of messages](/guides/pipelines/batching) into N batches, where each resulting batch contains a group of messages determined by a [Blob...

Splits a batch of messages into N batches, where each resulting batch contains a group of messages determined by a [function interpolated string](/gui...

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

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.

Executes a [JMESPath query](http://jmespath.org/) on JSON documents and replaces the message with the resulting document.

Transforms and filters messages using jq queries.

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

Prints a log event for each message. Messages always remain unchanged. The log message can be set using function interpolations described in [Bloblang...

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

Attaches Expanso runtime metadata to messages.

Emit custom metrics by extracting values from messages.

Converts messages to or from the [MessagePack](https://msgpack.org/) format.

Executes a [Bloblang](/guides/bloblang) mapping and directly transforms the contents of messages, mutating (or deleting) them.

Perform operations on a NATS key-value bucket.

nats_request_reply

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

Noop is a processor that does nothing, the message passes through unchanged. Why? Sometimes doing nothing is the braver option.

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

ollama_embeddings

Generates vector embeddings from text, using the Ollama API.

ollama_moderation

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

openai_chat_completion

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

openai_embeddings

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

openai_image_generation

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.

openai_transcription

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

openai_translation

Translates spoken audio into English, using the OpenAI API.

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

Converts batches of documents to or from [Parquet files](https://parquet.apache.org/docs/).

Decodes [Parquet files](https://parquet.apache.org/docs/) into a batch of structured messages.

Encodes [Parquet files](https://parquet.apache.org/docs/) from a batch of structured messages.

Parses common log formats into structured data. This is easier and often much faster than [`grok`](/components/processors/grok).

A processor grouping several sub-processors.

Performs conversions to or from a protobuf message. This processor uses reflection, meaning conversions can be made directly from the target .proto fi...

Query items within a [Qdrant](https://qdrant.tech/) collection.

Throttles the throughput of a pipeline according to a specified [`rate_limit`](/components/rate_limits/) resource. Rate limits are shared across compo...

Performs actions against Redis that aren't possible using a [`cache`](/components/processors/cache) processor. Actions are performed for each message ...

Performs actions against Redis using [LUA scripts](https://redis.io/docs/manual/programmability/eval-intro/).

Resource is a processor type that runs a processor resource identified by its label.

Attempts to execute a series of child processors until success.

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

Signs each message's payload with the edge node's Ed25519 identity keypair.

Sleep for a period of time specified as a duration string for each message. This processor will interpolate functions within the `duration` field, you...

Breaks message batches (synonymous with multiple part messages) into smaller batches. The size of the resulting batches are determined either by a dis...

Runs an arbitrary SQL query against a database and (optionally) returns the result as an array of objects, one for each row returned.

Inserts rows into an SQL database for each message, and leaves the message unchanged.

Runs an arbitrary SQL query against a database and (optionally) returns the result as an array of objects, one for each row returned.

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

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.

Conditionally processes messages based on their contents.

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.

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

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

Unarchives messages according to the selected archive format into multiple messages within a [batch](/guides/pipelines/batching).

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

Executes a topology of [`branch` processors](/components/processors/branch), 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.

Processor Pipelines
Error Handling
Browse Processors