redpanda_data_transform

Available in: Self-Managed

Executes a Redpanda Data Transform as a processor

Introduced in version 4.31.0.

  • Common

  • Advanced

# Common config fields, showing default values
label: ""
redpanda_data_transform:
  module_path: "" # No default (required)
  input_key: "" # No default (optional)
  output_key: "" # No default (optional)
  input_headers:
    include_prefixes: []
    include_patterns: []
  output_metadata:
    include_prefixes: []
    include_patterns: []
# All config fields, showing default values
label: ""
redpanda_data_transform:
  module_path: "" # No default (required)
  input_key: "" # No default (optional)
  output_key: "" # No default (optional)
  input_headers:
    include_prefixes: []
    include_patterns: []
  output_metadata:
    include_prefixes: []
    include_patterns: []
  timestamp_ms: ${! timestamp_unix_milli() } # No default (optional)
  timeout: 10s
  max_memory_pages: 1600

This processor executes a Redpanda Data Transform WebAssembly module, calling OnRecordWritten for each message being processed.

You can find out about how transforms work here: https://docs.redpanda.com/current/develop/data-transforms/how-transforms-work/

Fields

input_headers

Determine which (if any) metadata values should be added to messages as headers.

Type: object

input_headers.include_patterns[]

Provide a list of explicit metadata key regular expression (re2) patterns to match against.

Type: array

Default: []

# Examples:
include_patterns:
  - .*

  - _timestamp_unix$

input_headers.include_prefixes[]

Provide a list of explicit metadata key prefixes to match against.

Type: array

Default: []

# Examples:
include_prefixes:
  - foo_
  - bar_

  - kafka_

  - content-

input_key

An optional key to populate for each message. This field supports interpolation functions.

Type: string

max_memory_pages

The maximum amount of wasm memory pages (64KiB) that an individual wasm module instance can use

Type: int

Default: 1600

module_path

The path of the target WASM module to execute.

Type: string

output_key

An optional name of metadata for an output message key.

Type: string

output_metadata

Determine which (if any) message headers should be added to the output as metadata.

Type: object

output_metadata.include_patterns[]

Provide a list of explicit metadata key regular expression (re2) patterns to match against.

Type: array

Default: []

# Examples:
include_patterns:
  - .*

  - _timestamp_unix$

output_metadata.include_prefixes[]

Provide a list of explicit metadata key prefixes to match against.

Type: array

Default: []

# Examples:
include_prefixes:
  - foo_
  - bar_

  - kafka_

  - content-

timeout

The maximum period of time for a message to be processed

Type: string

Default: 10s

timestamp

An optional timestamp to set for each message. When left empty, the current timestamp is used.

Type: string

# Examples:
timestamp: ${! timestamp_unix() }
timestamp: ${! metadata("kafka_timestamp_ms") }