Connect
Also in Cloud

batched

Consumes data from a child input and applies a batching policy to the stream.

Introduced in version 4.11.0.

Batching at the input level is sometimes useful for processing across micro-batches, and can also sometimes be a useful performance trick. However, most inputs are fine without it so unless you have a specific plan for batching this component is not worth using.

  • Common

  • Advanced

input:
  label: ""
  batched:
    child: {} # No default (required)
    policy:
      count: 0
      byte_size: 0
      period: ""
      check: ""
input:
  label: ""
  batched:
    child: {} # No default (required)
    policy:
      count: 0
      byte_size: 0
      period: ""
      check: ""
      processors: [] # No default (optional)

Fields

child

The child input.

Type: input

policy

Configure a batching policy.

Type: object

# Examples:
policy:
  byte_size: 5000
  count: 0
  period: 1s

# ---

policy:
  count: 10
  period: 1s

# ---

policy:
  check: this.contains("END BATCH")
  count: 0
  period: 1m

policy.byte_size

The maximum total size (in bytes) that a batch can reach before it is flushed. When the combined size of all messages in the batch reaches or exceeds this limit, the batch is immediately sent to the next stage (such as a processor or output).

Set to 0 to disable size-based batching. When disabled, messages are flushed based on other conditions (such as count or period).

Type: int

Default: 0

policy.check

A Bloblang query that returns a boolean value indicating whether a message should end a batch.

Type: string

Default: ""

# Examples:
check: this.type == "end_of_transaction"

policy.count

The number of messages at which the batch is flushed. Set to 0 to disable count-based batching.

Type: int

Default: 0

policy.period

The length of time after which an incomplete batch is flushed regardless of its size. This field accepts Go duration format strings such as 100ms, 1s, or 5s. Supported time units are ns, us, ms, s, m, and h.

Type: string

Default: ""

# Examples:
period: 1s

# ---

period: 1m

# ---

period: 500ms

policy.processors[]

A list of processors to apply to a batch as it is flushed. This allows you to aggregate and archive the batch however you see fit. All resulting messages are flushed as a single batch, so splitting the batch into smaller batches with these processors has no effect.

Type: array<processor>

# Examples:
processors:
  - archive:
      format: concatenate


# ---

processors:
  - archive:
      format: lines


# ---

processors:
  - archive:
      format: json_array