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

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

aws_dynamodb_partiql

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

IntegrationAWSCloud+2

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

IntegrationAWSCloud

Creates or updates messages as JSON documents in Azure CosmosDB.

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.

MappingParsingTransform+1

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

CompositionTransformRouting

Performs operations against a cache resource 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]

Stops the current pipeline process with a fatal log message.

Forwards a numeric reading only when it has changed meaningfully since the last forwarded value. Drops sensor jitter, sub-significant drift, and flat-...

FilterTransform

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.

IntegrationGCPCloud

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 a text string, 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 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 evalua...

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

IntegrationNetworkHTTP

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

MappingParsingTransform

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

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

MappingParsingTransform+1

Attaches Expanso runtime metadata — pipeline IDs, node info, runtime counters, custom fields — to messages, either as message metadata or spliced into...

MappingCompositionCore

Emit custom metrics by extracting values from messages.

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

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

Perform operations on a NATS key-value bucket.

ServicesMessagingStreaming

nats_request_reply

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

ServicesMessagingStreaming

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 and external tools.

ollama_embeddings

Generates vector embeddings from text, using the Ollama API.

ollama_moderation

Checks the responses of a large language model against safety and security standards, using the Ollama API.

openai_chat_completion

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

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 the OpenAI API.

Generates audio from a text description and other attributes, using the 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.

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

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

ParsingTransform

Query items within a Qdrant collection and filter the returned results.

ServicesDatabase

Throttles the throughput of a pipeline according to a specified `rate_limit` resource. Rate limits are shared across components and therefore apply gl...

Performs actions against Redis that aren't possible using a `cache` processor. Actions are

IntegrationMessagingCaching

Performs actions against Redis using LUA scripts.

IntegrationMessagingCaching

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.

Deterministically signs each pipeline message with the edge node's Ed25519 identity. Adds signature, keyId, and algorithm to message metadata, the JSO...

SecurityTransform

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.

ServicesDatabase

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.

CompositionTransformRouting

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

Attempts a list of child processors and applies a separate list of catch processors if any of them fail.

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

Executes a topology of [`branch` processors][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.

ParsingTransform

Processor Pipelines
Error Handling
Browse Processors