---
title: Get data into Streams
description: Learn how to get data into Streams.
url: https://www.elastic.co/elastic/docs-builder/docs/3717/solutions/observability/streams/get-data-in
products:
  - Elastic Cloud Enterprise
  - Elastic Cloud Hosted
  - Elastic Cloud Serverless
  - Elastic Cloud on Kubernetes
  - Elastic Observability
  - Elastic Stack
  - Elasticsearch
  - Kibana
applies_to:
  - Elastic Cloud Serverless: Generally available
  - Elastic Stack: Generally available since 9.2, Preview in 9.1
---

# Get data into Streams
This page covers the different ways to send data to Streams:
- **[Ingest new data](#get-data-in-wired)**: Use wired streams to send logs to a managed endpoint for new ingestion. Data lands in a managed hierarchy with inheritance, partitioning, and cascading configuration.
  Best for new deployments, custom logs, and mixed-format sources.
- **[Work with existing data](#get-data-in-classic)**: Use classic streams to work with data already flowing into Elasticsearch. No migration or configuration changes required.


## Before you get started

Streams requires the following permissions:
<applies-switch>
  <applies-item title="serverless:" applies-to="Elastic Cloud Serverless: Generally available">
    Streams requires one of the following Elastic Cloud Serverless roles:
    - Admin: Able to manage all Streams
    - Editor/Viewer: Has limited access to Streams, cannot perform all actions
  </applies-item>

  <applies-item title="stack:" applies-to="Elastic Stack: Generally available">
    To manage all streams, you need the following permissions:
    - **Cluster permissions**: `manage_index_templates`, `manage_ingest_pipelines`, `manage_pipeline`, `read_pipeline`
    - **Data stream level permissions**: `read`, `write`, `create`, `manage`, `monitor`, `manage_data_stream_lifecycle`, `read_failure_store`, `manage_failure_store`, `manage_ilm`.
    To view streams, you need the following permissions:
    - **Data stream level**: `read`, `view_index_metadata`, `monitor`
    For more information, refer to [Cluster privileges](https://docs-v3-preview.elastic.dev/elastic/docs-builder/docs/3717/reference/elasticsearch/security-privileges#privileges-list-cluster) and [Granting privileges for data streams and aliases](https://www.elastic.co/elastic/docs-builder/docs/3717/deploy-manage/users-roles/cluster-or-deployment-auth/granting-privileges-for-data-streams-aliases).
  </applies-item>
</applies-switch>


## Ingest new data with wired streams

<applies-to>
  - Elastic Cloud Serverless: Preview
  - Elastic Stack: Preview since 9.2
</applies-to>

Wired streams send your documents to a managed endpoint, from which you can route data into child streams based on [partitioning](https://www.elastic.co/elastic/docs-builder/docs/3717/solutions/observability/streams/organize-your-data) rules. Child streams automatically inherit mappings, lifecycle settings, and processors from the parent, and configuration changes propagate through the hierarchy.
<applies-to>Elastic Cloud Serverless: Preview, Elastic Stack: Preview since 9.4</applies-to> Two endpoints are available. Use **`logs.otel`** (recommended) when sending OTel-native data or when you want a consistent, normalized format. Streams translates ECS field names to OTel equivalents automatically. Use **`logs.ecs`** when your data already uses ECS field names and you want to preserve them without transformation.
To send data to a wired stream, configure your shipper to point to the appropriate endpoint:
<tab-set>
  <tab-item title="OpenTelemetry">
    <note>
      Set the index based on your Elastic Stack version:
      - <applies-to>Elastic Cloud Serverless: Preview</applies-to> <applies-to>Elastic Stack: Preview since 9.4</applies-to> Set the index to `logs.otel` or `logs.ecs`, depending on the endpoint you want to use.
      - <applies-to>Elastic Stack: Preview from 9.2 to 9.3</applies-to> Set the index to `logs`. Only the `logs` endpoint is available in these versions.
    </note>

    ```yaml
    processors:
      transform/logs-streams:
        log_statements:
          - context: resource
            statements:
              - set(attributes["elasticsearch.index"], "logs.otel")
    service:
      pipelines:
        logs:
          receivers: [myreceiver]
          processors: [transform/logs-streams]
          exporters: [elasticsearch, otlp]
    ```
  </tab-item>

  <tab-item title="Filebeat">
    <note>
      Set the index based on your Elastic Stack version:
      - <applies-to>Elastic Cloud Serverless: Preview</applies-to> <applies-to>Elastic Stack: Preview since 9.4</applies-to> Set the index to `logs.otel` or `logs.ecs`, depending on the endpoint you want to use.
      - <applies-to>Elastic Stack: Preview from 9.2 to 9.3</applies-to> Set the index to `logs`. Only the `logs` endpoint is available in these versions.
    </note>

    ```yaml
    filebeat.inputs:
      - type: filestream
        id: my-filestream-id
        index: logs.otel
        enabled: true
        paths:
          - /var/log/*.log

    # No need to install templates for wired streams
    setup:
      template:
        enabled: false

    output.elasticsearch:
      hosts: ["<elasticsearch-host>"]
      api_key: "<your-api-key>"
    ```
  </tab-item>

  <tab-item title="Logstash">
    <note>
      Set the index based on your Elastic Stack version:
      - <applies-to>Elastic Cloud Serverless: Preview</applies-to> <applies-to>Elastic Stack: Preview since 9.4</applies-to> Set the index to `logs.otel` or `logs.ecs`, depending on the endpoint you want to use.
      - <applies-to>Elastic Stack: Preview from 9.2 to 9.3</applies-to> Set the index to `logs`. Only the `logs` endpoint is available in these versions.
    </note>

    ```json
    output {
      elasticsearch {
        hosts => ["<elasticsearch-host>"]
        api_key => "<your-api-key>"
        index => "logs.otel"
        action => "create"
      }
    }
    ```
  </tab-item>

  <tab-item title="Fleet">
    Use the **Custom Logs (Filestream)** integration to send data to wired streams:
    1. Find **Fleet** in the navigation menu or use the [global search field](https://www.elastic.co/elastic/docs-builder/docs/3717/explore-analyze/find-and-organize/find-apps-and-objects).
    2. Select the **Settings** tab.
    3. Under **Outputs**, find the output you want to use and select the `pencil` icon.
    4. Turn on **Write to logs streams**.
    5. Add the **Custom Logs (Filestream)** integration to an agent policy.
    6. Enable the **Use the "logs" data stream** setting under **Change defaults**.
    7. Under **Where to add this integration**, select an agent policy that uses the output configured in step 4.
  </tab-item>

  <tab-item title="API">
    <note>
      Set the endpoint based on your Elastic Stack version:
      - <applies-to>Elastic Cloud Serverless: Preview</applies-to> <applies-to>Elastic Stack: Preview since 9.4</applies-to> Set the endpoint to `logs.otel` or `logs.ecs`, depending on the endpoint you want to use.
      - <applies-to>Elastic Stack: Preview from 9.2 to 9.3</applies-to> Set the endpoint to `logs`. Only the `logs` endpoint is available in these versions.
    </note>
    Send data to the endpoint using the [Bulk API](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-bulk):
    ```json
    POST /logs.otel/_bulk
    { "create": {} }
    { "@timestamp": "2025-05-05T12:12:12", "body": { "text": "Hello world!" }, "resource": { "attributes": { "host.name": "my-host-name" } } }
    { "create": {} }
    { "@timestamp": "2025-05-05T12:12:12", "message": "Hello world!", "host.name": "my-host-name" }
    ```
  </tab-item>
</tab-set>


### Verify data is flowing

After configuring your data source, confirm data is appearing in Discover.
For wired streams, you first need to make the index pattern available:
1. Manually [create a data view](/elastic/docs-builder/docs/3717/explore-analyze/find-and-organize/data-views#settings-create-pattern) for the wired streams index pattern (`logs,logs.*`).
2. Add the wired streams index pattern (`logs,logs.*`) to the `observability:logSources` Kibana advanced setting, which you can open from the navigation menu or by using the [global search field](https://www.elastic.co/elastic/docs-builder/docs/3717/explore-analyze/find-and-organize/find-apps-and-objects).

Once data appears in Discover, you're ready to start organizing, parsing, and configuring retention for your streams.

#### Query unmapped fields

<applies-to>
  - Elastic Cloud Serverless: Preview
  - Elastic Stack: Preview since 9.4
</applies-to>

Wired streams can contain fields stored in `_source` that are not explicitly mapped. By default, ES|QL returns an error when a query references an unmapped field. To make unmapped fields queryable, add `SET unmapped_fields = "LOAD";` at the start of your ES|QL query:
```esql
SET unmapped_fields = "LOAD";
FROM logs.otel
| WHERE my_custom_field == "value"
```

When `LOAD` is set, unmapped fields are loaded from `_source` as `keyword` fields, or treated as null if absent from `_source`.
<applies-to>Elastic Stack: Planned</applies-to> When you query a wired stream and the ES|QL editor detects an unknown column error, a **Load unmapped fields** quick fix action is available. Select it to apply this setting automatically.
For a conceptual overview and use cases, refer to [Unmapped fields](https://docs-v3-preview.elastic.dev/elastic/docs-builder/docs/3717/reference/query-languages/esql/esql-unmapped-fields). For Kibana editor behavior, refer to [Handle unmapped fields with `SET unmapped_fields`](/elastic/docs-builder/docs/3717/explore-analyze/query-filter/languages/esql-kibana#esql-kibana-unmapped-fields).

## Work with existing data with classic streams

Classic streams let you use the Streams UI to extract fields and configure data retention for data that's already being ingested into Elasticsearch without additional configuration.
<warning>
  Do not reroute or migrate existing data streams to the `logs`, `logs.otel`, or `logs.ecs` endpoints. Classic streams work with your existing data streams in place. No migration is required.
</warning>

Classic streams:
- Are based on existing data streams, index templates, and component templates.
- Can follow the data retention policy set in the existing index template.
- Do not support hierarchical inheritance or cascading configuration updates.

Open classic streams from the following places in Kibana:
- Select **Streams** from the navigation menu or use the [global search field](https://www.elastic.co/elastic/docs-builder/docs/3717/explore-analyze/find-and-organize/find-apps-and-objects), and find the data stream in the Streams table.
- Open the data stream for a specific document from **Discover**. To do this:
  1. From the **Documents** table, select the `expand` icon for a document to expand the details flyout.
2. Under **Stream**, select the link to the stream (for example, `logs-generic.default`). The Streams UI opens filtered to the stream that contains the document.

<applies-to>Elastic Cloud Serverless: Preview</applies-to> <applies-to>Elastic Stack: Preview since 9.1</applies-to> You can also access Streams features using the Streams API. Refer to the [Streams API documentation](https://www.elastic.co/docs/api/doc/kibana/group/endpoint-streams) for more information.

## Next steps

Once your data is flowing into Streams, you can start organizing and enriching it:
- **[Organize your data](https://www.elastic.co/elastic/docs-builder/docs/3717/solutions/observability/streams/organize-your-data)**: Use partitioning to route data subsets into dedicated child streams with independent retention and processing rules. Partitioning is only available for wired streams.
- **[Parse and process](https://www.elastic.co/elastic/docs-builder/docs/3717/solutions/observability/streams/parse-and-process)**: Build a processing pipeline to extract structured fields from raw log messages using AI-generated or manually configured processors.