Process your documents

After selecting a stream, use the Processing tab to add processors and conditions that modify your unstructured documents and extract meaningful fields, so you can filter and analyze your data more effectively.

For example, in Discover, extracted fields might let you filter for log messages with an ERROR log level that occurred during a specific time period to help diagnose an issue. Without extracting the log level and timestamp fields from your messages, those filters wouldn't return meaningful results.

• Add processors: Use the Streams UI without needing to manually configuring configuring pipeline JSON or Grok syntax.
• Generate pipeline suggestions using AI: Let Streams analyze sample documents and suggests pipeline patterns, so you're refining instead of writing from scratch.
• Preview changes: Use the data preview to view which fields your pattern extracts per document, removing the guesswork.
• Detect and resolve processing issues: Identify which processor or condition is causing documents to fail during processing.
• Catch mapping conflicts: Identify potential mapping conflicts before they cause cluster-wide failures. Streams simulates the indexing process end-to-end before deploying.

Streams uses Elasticsearch ingest pipelines made up of processors to transform your data, without requiring you to switch interfaces and manually update pipelines.

To add a processor from the Processing tab:

To modify an existing processor, open the actions menu next to it to see the available options:

• Move up or Move down: Change the order of the processor.
• Add description: Change the processor description from its metadata to a description of your choice.
• Remove description: For processors with an added description, use this option to return the description to the metadata.
• Edit: Modify the processor configuration.
• Duplicate: Create another processor with the same configuration to use as a template.
• Delete: Remove the processor permanently.

The Streams processing UI provides an interactive mode and a YAML mode for editing processors and conditions.

To switch modes, select the appropriate tab from the top of the processing page.

×

Streams defaults to interactive mode unless the configuration can't be represented in interactive mode (for example, when nesting levels are too deep).

Interactive mode provides a form-based interface for creating and editing processors. This mode works best for:

• Users who prefer a guided, visual approach
• Configurations that don't require deeply nested conditions

YAML mode provides a code editor for writing Streamlang directly. This mode works best for:

• Users who prefer working with code
• Advanced configurations with complex or deeply nested conditions

Refer to the Streamlang reference for the complete syntax, condition operators, and examples.

When you save processors, Streams appends processing to the best-matching ingest pipeline for the data stream. It either chooses the best-matching pipeline ending in @custom in your data stream, or it adds one for you.

Streams identifies the appropriate @custom pipeline (for example, logs-myintegration@custom or logs@custom) by checking the default_pipeline that is set on the data stream. You can view the default pipeline on the Advanced tab under Ingest pipeline.

In this default pipeline, Streams locates the last processor that calls a pipeline ending in @custom.

• For integrations, this would result in a pipeline name like logs-myintegration@custom.
• Without an integration, the only @custom pipeline available may be logs@custom.

If no default pipeline is detected, Streams adds a default pipeline to the data stream by updating the index templates.

If a default pipeline is detected, but it does not contain a custom pipeline, Streams adds the pipeline processor directly to the pipeline.

Streams then adds a pipeline processor to the end of that @custom pipeline. This processor definition directs matching documents to a dedicated pipeline managed by Streams called <data_stream_name>@stream.processing:

// Example processor added to the relevant @custom pipeline
{
  "pipeline": {
    "name": "<data_stream_name>@stream.processing",
    "if": "ctx._index == '<data_stream_name>'",
    "ignore_missing_pipeline": true,
    "description": "Call the stream's managed pipeline - do not change this manually but instead use the Streams UI or API"
  }
}
		

1. for example, logs-my-app-default@stream.processing

Streams then creates and manages the <data_stream_name>@stream.processing pipeline, adding the processors you configured in the UI.

Do not manually modify the <data_stream_name>@stream.processing pipeline created by Streams. You can still add your own processors manually to the @custom pipeline if needed. Adding processors before the pipeline processor created by Streams might cause unexpected behavior.

• Streams does not support all processors. Refer to the Streamlang reference for supported processors.
• The data preview simulation might not accurately reflect the changes to the existing data when editing existing processors or re-ordering them.
• Streams can't properly handle arrays. While it supports basic actions like appending or renaming, it can't access individual array elements. For classic streams, the workaround is to use the manual pipeline configuration that supports Painless scripting and all ingest processors.