数据流

数据流（Data streams） #

如果你正在将连续生成的时间序列数据（如日志、事件和指标）摄入 Easysearch，那么你很可能处于这样一种场景：文档数量快速增长，且你无需更新旧文档。

管理时间序列数据的典型工作流程包含多个步骤，例如创建滚动索引别名、定义写入索引，以及为底层索引定义通用的映射和设置。

数据流简化了这一过程，并强制采用最适合时间序列数据的配置方式，例如主要为仅追加（append-only）数据设计，并确保每个文档都包含一个时间戳字段。

数据流在内部由多个底层索引组成。搜索请求会被路由到所有底层索引，而写入请求则被路由到最新的写入索引。通过 索引生命周期管理（ILM） 策略，你可以自动处理索引滚动（rollover）或删除操作。

数据流使用说明 #

步骤 1：创建索引模板 #

要创建数据流，首先需要创建一个索引模板，用于将一组索引配置为数据流。data_stream 对象表明这是一个数据流，而非普通索引模板。索引模式需与数据流的名称匹配：

PUT _index_template/logs-template-nginx
{
  "index_patterns": "logs-nginx",
  "data_stream": {
  },
  "priority": 200,
  "template": {
    "settings": {
      "number_of_shards": 1,
      "number_of_replicas": 0
    }
  }
}

在此情况下，每个摄入的文档都必须包含一个 @timestamp 字段。

你也可以在 data_stream 对象中自定义时间戳字段名称。此外，你还可以在此处定义索引映射和其他设置，就像为普通索引模板所做的那样。

PUT _index_template/logs-template-nginx
{
  "index_patterns": "logs-nginx",
  "data_stream": {
    "timestamp_field": {
      "name": "request_time"
    }
  },
  "priority": 200,
  "template": {
    "settings": {
      "number_of_shards": 1,
      "number_of_replicas": 0
    }
  }
}

在此示例中，logs-nginx 索引会匹配 logs-template-nginx 模板。当存在多个匹配时，Easysearch 会选择优先级更高的模板。

步骤 2：创建数据流 #

创建索引模板后，即可创建数据流。

你可以使用数据流 API 显式创建数据流。该 API 将初始化第一个底层索引：

PUT _data_stream/logs-nginx

你也可以直接开始摄入数据，而无需预先创建数据流。

由于我们已有包含 data_stream 对象的匹配索引模板，Easysearch 将自动创建数据流：

POST logs-nginx/_doc
{
  "message": "login attempt failed",
  "@timestamp": "2013-03-01T00:00:00"
}

要查看特定数据流的信息：

GET _data_stream/logs-nginx

示例响应 #

{
  "data_streams": [
    {
      "name": "logs-nginx",
      "timestamp_field": {
        "name": "@timestamp"
      },
      "indices": [
        {
          "index_name": ".ds-logs-nginx-000001",
          "index_uuid": "Z8RATqnzTbGxbm-khb5LKw"
        },
        {
          "index_name": ".ds-logs-nginx-000002",
          "index_uuid": "o8xka3CxSjqzNrw1BoRP6A"
        }
      ],
      "generation": 2,
      "status": "GREEN",
      "template": "logs-template-nginx"
    }
  ]
}

你可以看到时间戳字段的名称、底层索引列表、用于创建数据流的模板，以及数据流的健康状态（反映其所有底层索引中最低的状态）。

要获取更多关于数据流的详细信息，请使用 _stats 端点：

GET _data_stream/logs-nginx/_stats

示例响应 #

{
  "_shards": {
    "total": 2,
    "successful": 2,
    "failed": 0
  },
  "data_stream_count": 1,
  "backing_indices": 2,
  "total_store_size_bytes": 18125,
  "data_streams": [
    {
      "data_stream": "logs-nginx",
      "backing_indices": 2,
      "store_size_bytes": 18125,
      "maximum_timestamp": 1362096000000
    }
  ]
}

要查看所有数据流的信息，请使用以下请求：

GET _data_stream

步骤 3：向数据流中摄入数据 #

你可以使用常规的索引 API 将数据摄入数据流。请确保每个索引的文档都包含时间戳字段。如果尝试摄入没有时间戳字段的文档，将会报错。

POST logs-nginx/_doc
{
  "message": "login attempt",
  "@timestamp": "2013-03-01T00:00:00"
}

步骤 4：搜索数据流 #

你可以像搜索普通索引或索引别名一样搜索数据流。搜索操作将应用于所有底层索引（即数据流中的全部数据）。

GET logs-nginx/_search
{
  "query": {
    "match": {
      "message": "login"
    }
  }
}

示例响应 #

{
  "took" : 514,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 0.2876821,
    "hits" : [
      {
        "_index" : ".ds-logs-redis-000001",
        "_type" : "_doc",
        "_id" : "-rhVmXoBL6BAVWH3mMpC",
        "_score" : 0.2876821,
        "_source" : {
          "message" : "login attempt",
          "@timestamp" : "2013-03-01T00:00:00"
        }
      }
    ]
  }
}

步骤 5：滚动数据流 #

滚动操作会创建一个新的底层索引，并将其设置为数据流的新写入索引。

要对数据流执行手动滚动操作：

POST logs-nginx/_rollover

示例响应 #

{
  "acknowledged": true,
  "shards_acknowledged": true,
  "old_index": ".ds-logs-nginx-000001",
  "new_index": ".ds-logs-nginx-000002",
  "rolled_over": true,
  "dry_run": false,
  "conditions": {}
}

现在，如果你对 logs-nginx 数据流执行 GET 操作，你会看到其生成（generation） ID 从 1 增加到了 2。

你还可以设置一个 索引生命周期管理（ILM）策略 来自动执行数据流的滚动操作。ILM 策略在底层索引创建时即被应用。当你将策略关联到数据流时，它仅影响该数据流未来的底层索引。

此外，你无需提供 rollover_alias 设置，因为 ILM 策略会从底层索引中自动推断此信息。

步骤 7：删除数据流 #

删除操作会自动删除数据流的所有底层索引，然后删除数据流本身。

要删除一个数据流及其所有隐藏的底层索引：

DELETE _data_stream/<data_stream_name>

你可以使用通配符来删除多个数据流。

我们建议使用 ILM 策略来删除数据流中的数据。

©INFINI.LTD, All Rights Reserved.