OpenSearch/docs/reference/data-streams/use-a-data-stream.asciidoc

[role="xpack"]
[[use-a-data-stream]]
== Use a data stream

After you <<set-up-a-data-stream,set up a data stream>>, you can do
the following:

* <<add-documents-to-a-data-stream>>
* <<search-a-data-stream>>
* <<get-stats-for-a-data-stream>>
* <<manually-roll-over-a-data-stream>>
* <<open-closed-backing-indices>>
* <<reindex-with-a-data-stream>>
* <<update-docs-in-a-data-stream-by-query>>
* <<delete-docs-in-a-data-stream-by-query>>
* <<update-delete-docs-in-a-backing-index>>

////
[source,console]
----
PUT /_index_template/my-data-stream-template
{
  "index_patterns": [ "my-data-stream*" ],
  "data_stream": { }
}

PUT /_data_stream/my-data-stream

POST /my-data-stream/_rollover/

POST /my-data-stream/_rollover/

PUT /my-data-stream/_create/bfspvnIBr7VVZlfp2lqX?refresh=wait_for
{
  "@timestamp": "2020-12-07T11:06:07.000Z",
  "user": {
    "id": "yWIumJd7"
  },
  "message": "Login successful"
}

PUT /_data_stream/my-data-stream-alt
----
// TESTSETUP

[source,console]
----
DELETE /_data_stream/*

DELETE /_index_template/*
----
// TEARDOWN
////

[discrete]
[[add-documents-to-a-data-stream]]
=== Add documents to a data stream

To add an individual document, use the <<docs-index_,index API>>.
<<ingest,Ingest pipelines>> are supported.

[source,console]
----
POST /my-data-stream/_doc/
{
  "@timestamp": "2020-12-07T11:06:07.000Z",
  "user": {
    "id": "8a4f500d"
  },
  "message": "Login successful"
}
----

You cannot add new documents to a data stream using the index API's `PUT
/<target>/_doc/<_id>` request format. To specify a document ID, use the `PUT
/<target>/_create/<_id>` format instead. Only an
<<docs-index-api-op_type,`op_type`>> of `create` is supported.

To add multiple documents with a single request, use the <<docs-bulk,bulk API>>.
Only `create` actions are supported.

[source,console]
----
PUT /my-data-stream/_bulk?refresh
{"create":{ }}
{ "@timestamp": "2020-12-08T11:04:05.000Z", "user": { "id": "vlb44hny" }, "message": "Login attempt failed" }
{"create":{ }}
{ "@timestamp": "2020-12-08T11:06:07.000Z", "user": { "id": "8a4f500d" }, "message": "Login successful" }
{"create":{ }}
{ "@timestamp": "2020-12-09T11:07:08.000Z", "user": { "id": "l7gk7f82" }, "message": "Logout successful" }
----

[discrete]
[[search-a-data-stream]]
=== Search a data stream

The following search APIs support data streams:

* <<search-search, Search>>
* <<async-search, Async search>>
* <<search-multi-search, Multi search>>
* <<search-field-caps, Field capabilities>>
* <<eql-search-api, EQL search>>

[discrete]
[[get-stats-for-a-data-stream]]
=== Get statistics for a data stream

Use the <<data-stream-stats-api,data stream stats API>> to get
statistics for one or more data streams:

[source,console]
----
GET /_data_stream/my-data-stream/_stats?human=true
----

[discrete]
[[manually-roll-over-a-data-stream]]
=== Manually roll over a data stream

Use the <<indices-rollover-index,rollover API>> to manually
<<data-streams-rollover,roll over>> a data stream:

[source,console]
----
POST /my-data-stream/_rollover/
----

[discrete]
[[open-closed-backing-indices]]
=== Open closed backing indices

You cannot search a <<indices-close,closed>> backing index, even by searching
its data stream. You also cannot <<update-docs-in-a-data-stream-by-query,update>>
or <<delete-docs-in-a-data-stream-by-query,delete>> documents in a closed index.

To re-open a closed backing index, submit an <<indices-open-close,open
index API request>> directly to the index:

[source,console]
----
POST /.ds-my-data-stream-000001/_open/
----

To re-open all closed backing indices for a data stream, submit an open index
API request to the stream:

[source,console]
----
POST /my-data-stream/_open/
----

[discrete]
[[reindex-with-a-data-stream]]
=== Reindex with a data stream

Use the <<docs-reindex,reindex API>> to copy documents from an
existing index, index alias, or data stream to a data stream. Because data streams are
<<data-streams-append-only,append-only>>, a reindex into a data stream must use
an `op_type` of `create`. A reindex cannot update existing documents in a data
stream.

////
[source,console]
----
PUT /_bulk?refresh=wait_for
{"create":{"_index" : "archive_1"}}
{ "@timestamp": "2020-12-08T11:04:05.000Z" }
{"create":{"_index" : "archive_2"}}
{ "@timestamp": "2020-12-08T11:06:07.000Z" }
{"create":{"_index" : "archive_2"}}
{ "@timestamp": "2020-12-09T11:07:08.000Z" }
{"create":{"_index" : "archive_2"}}
{ "@timestamp": "2020-12-09T11:07:08.000Z" }

POST /_aliases
{
  "actions" : [
    { "add" : { "index" : "archive_1", "alias" : "archive" } },
    { "add" : { "index" : "archive_2", "alias" : "archive", "is_write_index" : true} }
  ]
}
----
////

[source,console]
----
POST /_reindex
{
  "source": {
    "index": "archive"
  },
  "dest": {
    "index": "my-data-stream",
    "op_type": "create"
  }
}
----
// TEST[continued]

[discrete]
[[update-docs-in-a-data-stream-by-query]]
=== Update documents in a data stream by query

Use the <<docs-update-by-query,update by query API>> to update documents in a
data stream that match a provided query:

[source,console]
----
POST /my-data-stream/_update_by_query
{
  "query": {
    "match": {
      "user.id": "l7gk7f82"
    }
  },
  "script": {
    "source": "ctx._source.user.id = params.new_id",
    "params": {
      "new_id": "XgdX0NoX"
    }
  }
}
----

[discrete]
[[delete-docs-in-a-data-stream-by-query]]
=== Delete documents in a data stream by query

Use the <<docs-delete-by-query,delete by query API>> to delete documents in a
data stream that match a provided query:

[source,console]
----
POST /my-data-stream/_delete_by_query
{
  "query": {
    "match": {
      "user.id": "vlb44hny"
    }
  }
}
----

[discrete]
[[update-delete-docs-in-a-backing-index]]
=== Update or delete documents in a backing index

If needed, you can update or delete documents in a data stream by sending
requests to the backing index containing the document. You'll need:

* The <<mapping-id-field,document ID>>
* The name of the backing index containing the document
* If updating the document, its <<optimistic-concurrency-control,sequence number
and primary term>>

To get this information, use a <<search-a-data-stream,search request>>:

[source,console]
----
GET /my-data-stream/_search
{
  "seq_no_primary_term": true,
  "query": {
    "match": {
      "user.id": "yWIumJd7"
    }
  }
}
----

Response:

[source,console-result]
----
{
  "took": 20,
  "timed_out": false,
  "_shards": {
    "total": 3,
    "successful": 3,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": {
      "value": 1,
      "relation": "eq"
    },
    "max_score": 0.2876821,
    "hits": [
      {
        "_index": ".ds-my-data-stream-000003",      <1>
        "_type": "_doc",
        "_id": "bfspvnIBr7VVZlfp2lqX",              <2>
        "_seq_no": 0,                               <3>
        "_primary_term": 1,                         <4>
        "_score": 0.2876821,
        "_source": {
          "@timestamp": "2020-12-07T11:06:07.000Z",
          "user": {
            "id": "yWIumJd7"
          },
          "message": "Login successful"
        }
      }
    ]
  }
}
----
// TESTRESPONSE[s/"took": 20/"took": $body.took/]
// TESTRESPONSE[s/"max_score": 0.2876821/"max_score": $body.hits.max_score/]
// TESTRESPONSE[s/"_score": 0.2876821/"_score": $body.hits.hits.0._score/]

<1> Backing index containing the matching document
<2> Document ID for the document
<3> Current sequence number for the document
<4> Primary term for the document

To update the document, use an <<docs-index_,index API>> request with valid
`if_seq_no` and `if_primary_term` arguments:

[source,console]
----
PUT /.ds-my-data-stream-000003/_doc/bfspvnIBr7VVZlfp2lqX?if_seq_no=0&if_primary_term=1
{
  "@timestamp": "2020-12-07T11:06:07.000Z",
  "user": {
    "id": "8a4f500d"
  },
  "message": "Login successful"
}
----

To delete the document, use the <<docs-delete,delete API>>:

[source,console]
----
DELETE /.ds-my-data-stream-000003/_doc/bfspvnIBr7VVZlfp2lqX
----

To delete or update multiple documents with a single request, use the
<<docs-bulk,bulk API>>'s `delete`, `index`, and `update` actions. For `index`
actions, include valid <<bulk-optimistic-concurrency-control,`if_seq_no` and
`if_primary_term`>> arguments.

[source,console]
----
PUT /_bulk?refresh
{ "index": { "_index": ".ds-my-data-stream-000003", "_id": "bfspvnIBr7VVZlfp2lqX", "if_seq_no": 0, "if_primary_term": 1 } }
{ "@timestamp": "2020-12-07T11:06:07.000Z", "user": { "id": "8a4f500d" }, "message": "Login successful" }
----