1
0

use-a-data-stream.asciidoc 17 KB

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