python

EXPERIMENTAL

This component is experimental and therefore subject to change or removal outside of major version releases.

Executes a Python script against each message in a stream.

# Config fields, showing default values
label: ""
python:
  script: "" # No default (required)
  imports: []

This processor uses a WebAssembly (WASM) hosted Python 3.12 environment provided by VMWare.

Each message is passed to the script as a global variable this. The script should assign the transformed result to the global variable root.

Libraries

A curated set of standard libraries is available. Additional modules can be specified via the imports field, provided they are available within the internal WASM runtime environment.

Fields

script

The Python script to execute for each message.

Type: string

imports

An optional list of Python modules to pre-import for the script.

Type: array
Default: []

Examples

Structured Mapping

If we have a stream of JSON documents containing user data, we can use Python to calculate a new field.

pipeline:
  processors:
    - python:
        script: |
          root["full_name"] = f"{this['first_name']} {this['last_name']}"
          root["age_next_year"] = this["age"] + 1

# In: {"first_name": "Richie", "last_name": "Ryan", "age": 35}
# Out: {"full_name": "Richie Ryan", "age_next_year": 36}

Data Filtering

By assigning None to root, you can effectively filter out messages based on complex logic.

pipeline:
  processors:
    - python:
        script: |
          if this.pop("status", None) == "active":
              root = this

# In: {"status": "active", "region": "us-east-1"}
# Out: {"region": "us-east-1"}

# In: {"status": "inactive", "region": "af-south-1"}
# Out: null # empty

External Imports

Using standard libraries to perform calculations.

pipeline:
  processors:
    - python:
        imports: [ math ]
        script: |
          root = this
          root["rounded_value"] = math.ceil(this["value"])

# In: {"value": 3.14158}
# Out: {"value": 3.14158, "rounded_value": 3}