Streamlang

Streamlang is a YAML domain-specific language (DSL) for defining stream processing and routing logic. Streamlang provides a consistent processing interface that can be converted to multiple execution targets, including Elasticsearch ingest pipelines and ES|QL. This allows processing to run at ingest time or query time without rewriting rules.

You can write Streamlang directly using the YAML editing mode in the Processing tab or the interactive mode which generates Streamlang behind the scenes.

A Streamlang configuration is a YAML document with a single top-level steps array. Each step is either an action block (processor) or a condition block:

steps:
  - action: <processor_type>
    # processor-specific parameters
  - action: <processor_type>
    # processor-specific parameters
    where:
      # optional condition
  - condition:
      field: <field_path>
      eq: <value>
      steps:
        - action: <processor_type>
          # nested processor

Steps run in order. Each processor transforms input documents, and passes results to the next step.

Processors are the building blocks of a Streamlang configuration. Each processor has an action field that specifies an operation to perform.

All processors support the following common options:

The following table lists all available processors. Refer to the individual processor pages for YAML parameters and examples.

Streams exposes a Streamlang configuration, but internally it relies on Elasticsearch ingest pipeline processors and ES|QL. Streamlang doesn't always have 1:1 parity with the ingest processors because it needs to support options that work in both ingest pipelines and ES|QL. In most cases, you won't need to worry about these details, but the underlying design decisions still affect the UI and available configuration options. The following are some limitations and inconsistencies when using Streamlang processors:

• Consistently typed fields: ES|QL requires one consistent type per column, so workflows that produce mixed types across documents won't transpile.
• Conversion of types: ES|QL and ingest pipelines accept different conversion combinations and strictness (especially for strings), so convert can behave differently across targets.
• Multi-value commands/functions: Fields can contain one or multiple values. ES|QL and ingest processors don't always handle these cases the same way. For example, Grok in ES|QL handles multiple values automatically, while the Grok processor does not.
• Conditional execution: ES|QL's enforced table shape limits conditional casting, parsing, and wildcard field operations that ingest pipelines can do per-document.
• Arrays of objects / flattening: Ingest pipelines preserve nested JSON arrays, while ES|QL flattens to columns, so operations like rename and delete on parent objects can differ or fail.

Conditions are Boolean expressions that control when processors run and how wired streams route data into partitions. They appear in where clauses on processors, in condition blocks, and in stream partitioning.

Each comparison condition specifies a field and an operator with a value:

Use range to match values within a numeric range. You can combine any of gt, gte, lt, and lte:

where:
  field: attributes.status_code
  range:
    gte: 200
    lt: 300

Use exists to check whether a field is present:

# Field must exist
where:
  field: attributes.user_id
  exists: true

# Field must not exist
where:
  field: attributes.temp
  exists: false

Combine conditions using and, or, and not:

# All conditions must be true
where:
  and:
    - field: attributes.env
      eq: production
    - field: attributes.level
      eq: error

# At least one condition must be true
where:
  or:
    - field: attributes.level
      eq: error
    - field: attributes.level
      eq: warn

# Negate a condition
where:
  not:
    field: attributes.path
    startsWith: "/internal"

When partitioning data into child streams, conditions use the previous operators to define how to route documents to a child stream.

For example, the following routes documents to a child stream when attributes.filepath equals Linux.log:

field: attributes.filepath
eq: Linux.log

To enter conditions in YAML format when configuring a partition, turn on the Syntax editor under Condition in the Partitioning tab.

Condition blocks group processors so that they run only when they meet a condition. Use a condition step with nested steps:

steps:
  - condition:
      field: attributes.env
      eq: production
      steps:
        - action: set
          to: attributes.is_prod
          value: true
        - action: remove
          from: attributes.debug_info

You can nest condition blocks for complex logic:

steps:
  - condition:
      field: attributes.source
      eq: webserver
      steps:
        - action: grok
          from: body.message
          patterns:
            - "%{IP:attributes.client_ip} %{WORD:attributes.method} %{URIPATHPARAM:attributes.path} %{NUMBER:attributes.status}"
        - condition:
            field: attributes.status
            gte: 500
            steps:
              - action: set
                to: attributes.alert_level
                value: critical