use-a-data-stream.asciidoc 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700
  1. [role="xpack"]
  2. [[use-a-data-stream]]
  3. == Use a data stream
  4. After you <<set-up-a-data-stream,set up a data stream>>, you can do
  5. the following:
  6. * <<add-documents-to-a-data-stream>>
  7. * <<search-a-data-stream>>
  8. * <<manually-roll-over-a-data-stream>>
  9. * <<open-closed-backing-indices>>
  10. * <<reindex-with-a-data-stream>>
  11. * <<update-delete-docs-in-a-data-stream>>
  12. * <<update-delete-docs-in-a-backing-index>>
  13. ////
  14. [source,console]
  15. ----
  16. PUT /_index_template/logs_data_stream
  17. {
  18. "index_patterns": [ "logs*" ],
  19. "data_stream": {
  20. "timestamp_field": "@timestamp"
  21. }
  22. }
  23. PUT /_data_stream/logs
  24. POST /logs/_rollover/
  25. POST /logs/_rollover/
  26. PUT /logs/_create/bfspvnIBr7VVZlfp2lqX?refresh=wait_for
  27. {
  28. "@timestamp": "2020-12-07T11:06:07.000Z",
  29. "user": {
  30. "id": "yWIumJd7"
  31. },
  32. "message": "Login successful"
  33. }
  34. PUT /_data_stream/logs_alt
  35. ----
  36. // TESTSETUP
  37. [source,console]
  38. ----
  39. DELETE /_data_stream/*
  40. DELETE /_index_template/*
  41. ----
  42. // TEARDOWN
  43. ////
  44. [discrete]
  45. [[add-documents-to-a-data-stream]]
  46. === Add documents to a data stream
  47. You can add documents to a data stream using the following requests:
  48. * An <<docs-index_,index API>> request with an
  49. <<docs-index-api-op_type,`op_type`>> set to `create`. Specify the data
  50. stream's name in place of an index name.
  51. +
  52. --
  53. NOTE: The `op_type` parameter defaults to `create` when adding new documents.
  54. .*Example: Index API request*
  55. [%collapsible]
  56. ====
  57. The following index API request adds a new document to the `logs` data
  58. stream.
  59. [source,console]
  60. ----
  61. POST /logs/_doc/
  62. {
  63. "@timestamp": "2020-12-07T11:06:07.000Z",
  64. "user": {
  65. "id": "8a4f500d"
  66. },
  67. "message": "Login successful"
  68. }
  69. ----
  70. ====
  71. IMPORTANT: You cannot add new documents to a data stream using the index API's
  72. `PUT /<target>/_doc/<_id>` request format. To specify a document ID, use the
  73. `PUT /<target>/_create/<_id>` format instead.
  74. --
  75. * A <<docs-bulk,bulk API>> request using the `create` action. Specify the data
  76. stream's name in place of an index name.
  77. +
  78. --
  79. NOTE: Data streams do not support other bulk actions, such as `index`.
  80. .*Example: Bulk API request*
  81. [%collapsible]
  82. ====
  83. The following bulk API request adds several new documents to
  84. the `logs` data stream. Note that only the `create` action is used.
  85. [source,console]
  86. ----
  87. PUT /logs/_bulk?refresh
  88. {"create":{ }}
  89. { "@timestamp": "2020-12-08T11:04:05.000Z", "user": { "id": "vlb44hny" }, "message": "Login attempt failed" }
  90. {"create":{ }}
  91. { "@timestamp": "2020-12-08T11:06:07.000Z", "user": { "id": "8a4f500d" }, "message": "Login successful" }
  92. {"create":{ }}
  93. { "@timestamp": "2020-12-09T11:07:08.000Z", "user": { "id": "l7gk7f82" }, "message": "Logout successful" }
  94. ----
  95. ====
  96. --
  97. You can use an <<ingest,ingest pipeline>> with these requests to pre-process
  98. data before it's indexed.
  99. .*Example: Ingest pipeline*
  100. [%collapsible]
  101. ====
  102. The following <<put-pipeline-api,put pipeline API>> request creates the
  103. `lowercase_message_field` ingest pipeline. The pipeline uses the
  104. <<lowercase-processor,`lowercase` ingest processor>> to change the `message`
  105. field value to lowercase before indexing.
  106. [source,console]
  107. ----
  108. PUT /_ingest/pipeline/lowercase_message_field
  109. {
  110. "description" : "Lowercases the message field value",
  111. "processors" : [
  112. {
  113. "lowercase" : {
  114. "field" : "message"
  115. }
  116. }
  117. ]
  118. }
  119. ----
  120. // TEST[continued]
  121. The following index API request adds a new document to the `logs` data stream.
  122. The request includes a `?pipeline=lowercase_message_field` query parameter.
  123. This parameter indicates {es} should use the `lowercase_message_field` pipeline
  124. to pre-process the document before indexing it.
  125. During pre-processing, the pipeline changes the letter case of the document's
  126. `message` field value from `LOGIN Successful` to `login successful`.
  127. [source,console]
  128. ----
  129. POST /logs/_doc?pipeline=lowercase_message_field
  130. {
  131. "@timestamp": "2020-12-08T11:12:01.000Z",
  132. "user": {
  133. "id": "I1YBEOxJ"
  134. },
  135. "message": "LOGIN Successful"
  136. }
  137. ----
  138. // TEST[continued]
  139. ////
  140. [source,console]
  141. ----
  142. DELETE /_ingest/pipeline/lowercase_message_field
  143. ----
  144. // TEST[continued]
  145. ////
  146. ====
  147. [discrete]
  148. [[search-a-data-stream]]
  149. === Search a data stream
  150. The following search APIs support data streams:
  151. * <<search-search, Search>>
  152. * <<async-search, Async search>>
  153. * <<search-multi-search, Multi search>>
  154. * <<search-field-caps, Field capabilities>>
  155. ////
  156. * <<eql-search-api, EQL search>>
  157. ////
  158. .*Example*
  159. [%collapsible]
  160. ====
  161. The following <<search-search,search API>> request searches the `logs` data
  162. stream for documents with a timestamp between today and yesterday that also have
  163. `message` value of `login successful`.
  164. [source,console]
  165. ----
  166. GET /logs/_search
  167. {
  168. "query": {
  169. "bool": {
  170. "must": {
  171. "range": {
  172. "@timestamp": {
  173. "gte": "now-1d/d",
  174. "lt": "now/d"
  175. }
  176. }
  177. },
  178. "should": {
  179. "match": {
  180. "message": "login successful"
  181. }
  182. }
  183. }
  184. }
  185. }
  186. ----
  187. ====
  188. You can use a comma-separated list or wildcard (`*`) expression to search
  189. multiple data streams, indices, and index aliases in the same request.
  190. .*Example*
  191. [%collapsible]
  192. ====
  193. The following request searches the `logs` and `logs_alt` data streams, which are
  194. specified as a comma-separated list in the request path.
  195. [source,console]
  196. ----
  197. GET /logs,logs_alt/_search
  198. {
  199. "query": {
  200. "match": {
  201. "user.id": "8a4f500d"
  202. }
  203. }
  204. }
  205. ----
  206. The following request uses the `logs*` wildcard expression to search any data
  207. stream, index, or index alias beginning with `logs`.
  208. [source,console]
  209. ----
  210. GET /logs*/_search
  211. {
  212. "query": {
  213. "match": {
  214. "user.id": "vlb44hny"
  215. }
  216. }
  217. }
  218. ----
  219. The following search request omits a target in the request path. The request
  220. searches all data streams and indices in the cluster.
  221. [source,console]
  222. ----
  223. GET /_search
  224. {
  225. "query": {
  226. "match": {
  227. "user.id": "l7gk7f82"
  228. }
  229. }
  230. }
  231. ----
  232. ====
  233. [discrete]
  234. [[manually-roll-over-a-data-stream]]
  235. === Manually roll over a data stream
  236. A rollover creates a new backing index for a data stream. This new backing index
  237. becomes the stream's <<data-stream-write-index,write index>> and increments
  238. the stream's <<data-streams-generation,generation>>.
  239. In most cases, we recommend using <<index-lifecycle-management,{ilm-init}>> to
  240. automate rollovers for data streams. This lets you automatically roll over the
  241. current write index when it meets specified criteria, such as a maximum age or
  242. size.
  243. However, you can also use the <<indices-rollover-index,rollover API>> to
  244. manually perform a rollover. This can be useful if you want to
  245. <<data-streams-change-mappings-and-settings,apply mapping or setting changes>>
  246. to the stream's write index after updating a data stream's template.
  247. .*Example*
  248. [%collapsible]
  249. ====
  250. The following <<indices-rollover-index,rollover API>> request submits a manual
  251. rollover request for the `logs` data stream.
  252. [source,console]
  253. ----
  254. POST /logs/_rollover/
  255. ----
  256. ====
  257. [discrete]
  258. [[open-closed-backing-indices]]
  259. === Open closed backing indices
  260. You may <<indices-close,close>> one or more of a data stream's backing indices
  261. as part of its {ilm-init} lifecycle or another workflow. A closed backing index
  262. cannot be searched, even for searches targeting its data stream. You also can't
  263. <<update-delete-docs-in-a-data-stream,update or delete documents>> in a closed
  264. index.
  265. You can re-open individual backing indices by sending an
  266. <<indices-open-close,open request>> directly to the index.
  267. You also can conveniently re-open all closed backing indices for a data stream
  268. by sending an open request directly to the stream.
  269. .*Example*
  270. [%collapsible]
  271. ====
  272. The following <<cat-indices,cat indices>> API request retrieves the status for
  273. the `logs` data stream's backing indices.
  274. ////
  275. [source,console]
  276. ----
  277. POST /.ds-logs-000001,.ds-logs-000002/_close/
  278. ----
  279. ////
  280. [source,console]
  281. ----
  282. GET /_cat/indices/logs?v&s=index&h=index,status
  283. ----
  284. // TEST[continued]
  285. The API returns the following response. The response indicates the `logs` data
  286. stream contains two closed backing indices: `.ds-logs-000001` and
  287. `.ds-logs-000002`.
  288. [source,txt]
  289. ----
  290. index status
  291. .ds-logs-000001 close
  292. .ds-logs-000002 close
  293. .ds-logs-000003 open
  294. ----
  295. // TESTRESPONSE[non_json]
  296. The following <<indices-open-close,open API>> request re-opens any closed
  297. backing indices for the `logs` data stream, including `.ds-logs-000001` and
  298. `.ds-logs-000002`.
  299. [source,console]
  300. ----
  301. POST /logs/_open/
  302. ----
  303. // TEST[continued]
  304. You can resubmit the original cat indices API request to verify the
  305. `.ds-logs-000001` and `.ds-logs-000002` backing indices were re-opened.
  306. [source,console]
  307. ----
  308. GET /_cat/indices/logs?v&s=index&h=index,status
  309. ----
  310. // TEST[continued]
  311. The API returns the following response.
  312. [source,txt]
  313. ----
  314. index status
  315. .ds-logs-000001 open
  316. .ds-logs-000002 open
  317. .ds-logs-000003 open
  318. ----
  319. // TESTRESPONSE[non_json]
  320. ====
  321. [discrete]
  322. [[reindex-with-a-data-stream]]
  323. === Reindex with a data stream
  324. You can use the <<docs-reindex,reindex API>> to copy documents to a data stream
  325. from an existing index, index alias, or data stream.
  326. A reindex copies documents from a _source_ to a _destination_. The source and
  327. destination can be any pre-existing index, index alias, or data stream. However,
  328. the source and destination must be different. You cannot reindex a data stream
  329. into itself.
  330. Because data streams are <<data-streams-append-only,append-only>>, a reindex
  331. request to a data stream destination must have an `op_type` of `create`. This
  332. means a reindex can only add new documents to a data stream. It cannot update
  333. existing documents in the data stream destination.
  334. A reindex can be used to:
  335. * Convert an existing index alias and collection of time-based indices into a
  336. data stream.
  337. * Apply a new or updated <<create-a-data-stream-template,index template>>
  338. by reindexing an existing data stream into a new one. This applies mapping
  339. and setting changes in the template to each document and backing index of the
  340. data stream destination. See
  341. <<data-streams-use-reindex-to-change-mappings-settings>>.
  342. TIP: If you only want to update the mappings or settings of a data stream's
  343. write index, we recommend you update the <<create-a-data-stream-template,data
  344. stream's template>> and perform a <<manually-roll-over-a-data-stream,rollover>>.
  345. .*Example*
  346. [%collapsible]
  347. ====
  348. The following reindex request copies documents from the `archive` index alias to
  349. the existing `logs` data stream. Because the destination is a data stream, the
  350. request's `op_type` is `create`.
  351. ////
  352. [source,console]
  353. ----
  354. PUT /_bulk?refresh=wait_for
  355. {"create":{"_index" : "archive_1"}}
  356. { "@timestamp": "2020-12-08T11:04:05.000Z" }
  357. {"create":{"_index" : "archive_2"}}
  358. { "@timestamp": "2020-12-08T11:06:07.000Z" }
  359. {"create":{"_index" : "archive_2"}}
  360. { "@timestamp": "2020-12-09T11:07:08.000Z" }
  361. {"create":{"_index" : "archive_2"}}
  362. { "@timestamp": "2020-12-09T11:07:08.000Z" }
  363. POST /_aliases
  364. {
  365. "actions" : [
  366. { "add" : { "index" : "archive_1", "alias" : "archive" } },
  367. { "add" : { "index" : "archive_2", "alias" : "archive", "is_write_index" : true} }
  368. ]
  369. }
  370. ----
  371. ////
  372. [source,console]
  373. ----
  374. POST /_reindex
  375. {
  376. "source": {
  377. "index": "archive"
  378. },
  379. "dest": {
  380. "index": "logs",
  381. "op_type": "create"
  382. }
  383. }
  384. ----
  385. // TEST[continued]
  386. ====
  387. You can also reindex documents from a data stream to an index, index
  388. alias, or data stream.
  389. .*Example*
  390. [%collapsible]
  391. ====
  392. The following reindex request copies documents from the `logs` data stream
  393. to the existing `archive` index alias. Because the destination is not a data
  394. stream, the `op_type` does not need to be specified.
  395. [source,console]
  396. ----
  397. POST /_reindex
  398. {
  399. "source": {
  400. "index": "logs"
  401. },
  402. "dest": {
  403. "index": "archive"
  404. }
  405. }
  406. ----
  407. // TEST[continued]
  408. ====
  409. [discrete]
  410. [[update-delete-docs-in-a-data-stream]]
  411. === Update or delete documents in a data stream
  412. You can update or delete documents in a data stream using the following
  413. requests:
  414. * An <<docs-update-by-query,update by query API>> request
  415. +
  416. .*Example*
  417. [%collapsible]
  418. ====
  419. The following update by query API request updates documents in the `logs` data
  420. stream with a `user.id` of `l7gk7f82`. The request uses a
  421. <<modules-scripting-using,script>> to assign matching documents a new `user.id`
  422. value of `XgdX0NoX`.
  423. [source,console]
  424. ----
  425. POST /logs/_update_by_query
  426. {
  427. "query": {
  428. "match": {
  429. "user.id": "l7gk7f82"
  430. }
  431. },
  432. "script": {
  433. "source": "ctx._source.user.id = params.new_id",
  434. "params": {
  435. "new_id": "XgdX0NoX"
  436. }
  437. }
  438. }
  439. ----
  440. ====
  441. * A <<docs-delete-by-query,delete by query API>> request
  442. +
  443. .*Example*
  444. [%collapsible]
  445. ====
  446. The following delete by query API request deletes documents in the `logs` data
  447. stream with a `user.id` of `vlb44hny`.
  448. [source,console]
  449. ----
  450. POST /logs/_delete_by_query
  451. {
  452. "query": {
  453. "match": {
  454. "user.id": "vlb44hny"
  455. }
  456. }
  457. }
  458. ----
  459. ====
  460. [discrete]
  461. [[update-delete-docs-in-a-backing-index]]
  462. === Update or delete documents in a backing index
  463. Alternatively, you can update or delete documents in a data stream by sending
  464. the update or deletion request to the backing index containing the document. To
  465. do this, you first need to get:
  466. * The <<mapping-id-field,document ID>>
  467. * The name of the backing index that contains the document
  468. If you want to update a document, you must also get its current
  469. <<optimistic-concurrency-control,sequence number and primary term>>.
  470. You can use a <<search-a-data-stream,search request>> to retrieve this
  471. information.
  472. .*Example*
  473. [%collapsible]
  474. ====
  475. The following search request retrieves documents in the `logs` data stream with
  476. a `user.id` of `yWIumJd7`. By default, this search returns the document ID and
  477. backing index for any matching documents.
  478. The request includes a `"seq_no_primary_term": true` argument. This means the
  479. search also returns the sequence number and primary term for any matching
  480. documents.
  481. [source,console]
  482. ----
  483. GET /logs/_search
  484. {
  485. "seq_no_primary_term": true,
  486. "query": {
  487. "match": {
  488. "user.id": "yWIumJd7"
  489. }
  490. }
  491. }
  492. ----
  493. The API returns the following response. The `hits.hits` property contains
  494. information for any documents matching the search.
  495. [source,console-result]
  496. ----
  497. {
  498. "took": 20,
  499. "timed_out": false,
  500. "_shards": {
  501. "total": 3,
  502. "successful": 3,
  503. "skipped": 0,
  504. "failed": 0
  505. },
  506. "hits": {
  507. "total": {
  508. "value": 1,
  509. "relation": "eq"
  510. },
  511. "max_score": 0.2876821,
  512. "hits": [
  513. {
  514. "_index": ".ds-logs-000003", <1>
  515. "_id": "bfspvnIBr7VVZlfp2lqX", <2>
  516. "_seq_no": 0, <3>
  517. "_primary_term": 1, <4>
  518. "_score": 0.2876821,
  519. "_source": {
  520. "@timestamp": "2020-12-07T11:06:07.000Z",
  521. "user": {
  522. "id": "yWIumJd7"
  523. },
  524. "message": "Login successful"
  525. }
  526. }
  527. ]
  528. }
  529. }
  530. ----
  531. // TESTRESPONSE[s/"took": 20/"took": $body.took/]
  532. // TESTRESPONSE[s/"max_score": 0.2876821/"max_score": $body.hits.max_score/]
  533. // TESTRESPONSE[s/"_score": 0.2876821/"_score": $body.hits.hits.0._score/]
  534. <1> Backing index containing the matching document
  535. <2> Document ID for the document
  536. <3> Current sequence number for the document
  537. <4> Primary term for the document
  538. ====
  539. You can use an <<docs-index_,index API>> request to update an individual
  540. document. To prevent an accidental overwrite, this request must include valid
  541. `if_seq_no` and `if_primary_term` arguments.
  542. .*Example*
  543. [%collapsible]
  544. ====
  545. The following index API request updates an existing document in the `logs` data
  546. stream. The request targets document ID `bfspvnIBr7VVZlfp2lqX` in the
  547. `.ds-logs-000003` backing index.
  548. The request also includes the current sequence number and primary term in the
  549. respective `if_seq_no` and `if_primary_term` query parameters. The request body
  550. contains a new JSON source for the document.
  551. [source,console]
  552. ----
  553. PUT /.ds-logs-000003/_doc/bfspvnIBr7VVZlfp2lqX?if_seq_no=0&if_primary_term=1
  554. {
  555. "@timestamp": "2020-12-07T11:06:07.000Z",
  556. "user": {
  557. "id": "8a4f500d"
  558. },
  559. "message": "Login successful"
  560. }
  561. ----
  562. ====
  563. You use the <<docs-delete,delete API>> to delete individual documents. Deletion
  564. requests do not require a sequence number or primary term.
  565. .*Example*
  566. [%collapsible]
  567. ====
  568. The following index API request deletes an existing document in the `logs` data
  569. stream. The request targets document ID `bfspvnIBr7VVZlfp2lqX` in the
  570. `.ds-logs-000003` backing index.
  571. [source,console]
  572. ----
  573. DELETE /.ds-logs-000003/_doc/bfspvnIBr7VVZlfp2lqX
  574. ----
  575. ====
  576. You can use the <<docs-bulk,bulk API>> to delete or update multiple documents in
  577. one request using `delete`, `index`, or `update` actions.
  578. If the action type is `index`, the action must include valid
  579. <<bulk-optimistic-concurrency-control,`if_seq_no` and `if_primary_term`>>
  580. arguments.
  581. .*Example*
  582. [%collapsible]
  583. ====
  584. The following bulk API request uses an `index` action to update an existing
  585. document in the `logs` data stream.
  586. The `index` action targets document ID `bfspvnIBr7VVZlfp2lqX` in the
  587. `.ds-logs-000003` backing index. The action also includes the current sequence
  588. number and primary term in the respective `if_seq_no` and `if_primary_term`
  589. parameters.
  590. [source,console]
  591. ----
  592. PUT /_bulk?refresh
  593. { "index": { "_index": ".ds-logs-000003", "_id": "bfspvnIBr7VVZlfp2lqX", "if_seq_no": 0, "if_primary_term": 1 } }
  594. { "@timestamp": "2020-12-07T11:06:07.000Z", "user": { "id": "8a4f500d" }, "message": "Login successful" }
  595. ----
  596. ====