---
title: Get data into Streams
description: Learn how to get data into Streams.
url: https://docs-v3-preview.elastic.dev/elastic/docs-content/pull/8505/solutions/observability/streams/get-data-in
products:
  - Elastic Cloud Enterprise
  - Elastic Cloud Hosted
  - Elastic Cloud Serverless
  - Elastic Cloud on Kubernetes
  - Elastic Documentation
  - 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 shows you how to send new data to Streams or use Streams to work with existing data:
- **[Ingest new data](#get-data-in-wired)**: Use wired streams to send logs to a managed endpoint. This option suits 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 are required.


## Before you begin

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/elasticsearch/tree/main/reference/elasticsearch/security-privileges#privileges-list-cluster) and [Granting privileges for data streams and aliases](https://docs-v3-preview.elastic.dev/elastic/docs-content/pull/8505/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 documents to a managed endpoint, from which you can route data into child streams based on [partitioning](https://docs-v3-preview.elastic.dev/elastic/docs-content/pull/8505/solutions/observability/streams/organize-your-data) rules. Child streams inherit mappings, lifecycle settings, and processors from the parent.
Select the endpoint that matches your deployment:
- <applies-to>Elastic Cloud Serverless: Preview</applies-to> <applies-to>Elastic Stack: Preview since 9.4</applies-to> Use **`logs.otel`** for OTel-native data or for 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.
- <applies-to>Elastic Stack: Preview from 9.2 to 9.3</applies-to> Use **`logs`**. This is the only wired-stream endpoint available in these versions. Replace `logs.otel` with `logs` in the following examples.

Configure your shipper to use the endpoint:
<tab-set>
  <tab-item title="OpenTelemetry">
    ```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">
    ```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">
    ```txt
    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://docs-v3-preview.elastic.dev/elastic/docs-content/pull/8505/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">
    Send data to the endpoint using the [Bulk API](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-bulk):
    ```json

    { "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](https://docs-v3-preview.elastic.dev/elastic/docs-content/pull/8505/explore-analyze/find-and-organize/data-views/create-data-view) 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://docs-v3-preview.elastic.dev/elastic/docs-content/pull/8505/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: Preview since 9.5</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/elasticsearch/tree/main/reference/query-languages/esql/esql-unmapped-fields). For Kibana editor behavior, refer to [Handle unmapped fields with `SET unmapped_fields`](/elastic/docs-content/pull/8505/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://docs-v3-preview.elastic.dev/elastic/docs-content/pull/8505/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://docs-v3-preview.elastic.dev/elastic/docs-content/pull/8505/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://docs-v3-preview.elastic.dev/elastic/docs-content/pull/8505/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.