使用数据流
在您设置数据流后，您可以执行以下操作
要添加单个文档，请使用 index API。支持 摄入管道 (Ingest pipelines)。
POST /my-data-stream/_doc/
{
"@timestamp": "2099-03-08T11:06:07.000Z",
"user": {
"id": "8a4f500d"
},
"message": "Login successful"
}
您不能使用 index API 的 PUT /<target>/_doc/<_id> 请求格式向数据流添加新文档。要指定文档 ID，请改用 PUT /<target>/_create/<_id> 格式。仅支持 create 类型的 op_type。
要通过单个请求添加多个文档，请使用 bulk API。仅支持 create 操作。
PUT /my-data-stream/_bulk?refresh
{"create":{ }}
{ "@timestamp": "2099-03-08T11:04:05.000Z", "user": { "id": "vlb44hny" }, "message": "Login attempt failed" }
{"create":{ }}
{ "@timestamp": "2099-03-08T11:06:07.000Z", "user": { "id": "8a4f500d" }, "message": "Login successful" }
{"create":{ }}
{ "@timestamp": "2099-03-09T11:07:08.000Z", "user": { "id": "l7gk7f82" }, "message": "Logout successful" }
以下搜索 API 支持数据流
使用 data stream stats API 获取一个或多个数据流的统计信息
GET /_data_stream/my-data-stream/_stats?human=true
使用 rollover API 手动滚动数据流。手动滚动时有两个选项
立即触发滚动
POST /my-data-stream/_rollover/或者将滚动推迟到下一次索引事件发生时
POST /my-data-stream/_rollover?lazy使用第二个选项可以避免在不经常更新的数据流中出现空的后备索引。
您无法搜索已关闭的后备索引，即使通过搜索其数据流也无法做到。您也无法更新或删除已关闭索引中的文档。
要重新打开已关闭的后备索引，请直接向该索引提交 open index API 请求
POST /.ds-my-data-stream-2099.03.07-000001/_open/
要重新打开数据流的所有已关闭后备索引，请向该流提交 open index API 请求
POST /my-data-stream/_open/
使用 reindex API 将文档从现有索引、别名或数据流复制到数据流。由于数据流是仅追加 (append-only)的，因此重新索引到数据流中必须使用 create 类型的 op_type。重建索引无法更新数据流中的现有文档。
POST /_reindex
{
"source": {
"index": "archive"
},
"dest": {
"index": "my-data-stream",
"op_type": "create"
}
}
使用 update by query API 更新数据流中匹配指定查询的文档
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"
}
}
}
使用 delete by query API 删除数据流中匹配指定查询的文档
POST /my-data-stream/_delete_by_query
{
"query": {
"match": {
"user.id": "vlb44hny"
}
}
}
如果需要，您可以通过向包含文档的后备索引发送请求来更新或删除数据流中的文档。您将需要
- 文档 ID
- 包含文档的后备索引名称
- 如果要更新文档，需要其序列号和主要项 (sequence number and primary term)
要获取此信息，请使用搜索请求
GET /my-data-stream/_search
{
"seq_no_primary_term": true,
"query": {
"match": {
"user.id": "yWIumJd7"
}
}
}
响应
{
"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-2099.03.08-000003",
"_id": "bfspvnIBr7VVZlfp2lqX",
"_seq_no": 0,
"_primary_term": 1,
"_score": 0.2876821,
"_source": {
"@timestamp": "2099-03-08T11:06:07.000Z",
"user": {
"id": "yWIumJd7"
},
"message": "Login successful"
}
}
]
}
}
- 包含匹配文档的后备索引
- 文档的文档 ID
- 文档的当前序列号
- 文档的主要项
要更新文档，请使用带有有效 if_seq_no 和 if_primary_term 参数的 index API 请求
PUT /.ds-my-data-stream-2099-03-08-000003/_doc/bfspvnIBr7VVZlfp2lqX?if_seq_no=0&if_primary_term=1
{
"@timestamp": "2099-03-08T11:06:07.000Z",
"user": {
"id": "8a4f500d"
},
"message": "Login successful"
}
要删除文档，请使用 delete API
DELETE /.ds-my-data-stream-2099.03.08-000003/_doc/bfspvnIBr7VVZlfp2lqX
要通过单个请求删除或更新多个文档，请使用 bulk API 的 delete、index 和 update 操作。对于 index 操作，请包含有效的 if_seq_no 和 if_primary_term 参数。
PUT /_bulk?refresh
{ "index": { "_index": ".ds-my-data-stream-2099.03.08-000003", "_id": "bfspvnIBr7VVZlfp2lqX", "if_seq_no": 0, "if_primary_term": 1 } }
{ "@timestamp": "2099-03-08T11:06:07.000Z", "user": { "id": "8a4f500d" }, "message": "Login successful" }