Data stream-Elastic Stack 實戰手冊

作者：趙凱

Data stream 的概念

時序性數據

時間序列數據（ time series data ）是在不同時間上收集到的數據，用於所描述現象隨時間變化的情況。這類數據反映了某一事物、現象等，隨時間的變化狀態或程度。

總的來說，這類數據主要基於時間特性明顯，隨著時間的流逝，往往過去時間的數據沒有現在時間的重要或者敏感。

對於 Elastisearch 處理時序性數據，有人總結了主要有以下特點：

• 由時間戳 + 數據組成。基於時間的事件，可以是服務器日誌或者社交媒體流。
• 通常搜索最近事件，舊文件變得不太重要。
• 索引的使用主要基於時間，但是數據並不一定隨著時間均衡分佈。
• 時序性數據一旦存入後很少修改。
• 時序性數據隨著時間的增加，數據量會很大。

Elastisearch 在時序性數據的使用中，往往會有以下的缺點：

• 索引隨著時間增加而數目較多。
• 索引大小無法均衡。
• 管理索引成本較高，需要維護 merge 合併刪除等一系列任務。
• 節點資源與冷熱數據分佈不匹配。

在這樣的一個場景下，數據流 Data stream 應運而生。

Data stream （數據流）是 Elastic Stack 7.9 的一個新的功能。Data stream 可以跨多個索引存儲只追加時序性數據，同時為查詢寫入等請求提供唯一的一個命名資源。Data stream 非常適合日誌，事件，指標以及其他持續生成的數據。

簡單來說，Data stream 根據模板生成存儲數據的後備索引，然後自動將搜索或者索引請求路由到存儲流數據的後備索引。而這些後備索引則根據索引生命週期管理（ ILM ）來自動管理。

例如，你可以使用 ILM 自動將較舊的後備索引移動到較便宜的硬件上（冷熱數據處理），根據索引大小自動 Rollover 出新的後備索引，或者刪除到時間限制的索引。

在一定程度上，Data stream 的管理優勢是利用了 ILM 的特性。但是 ILM 在普通場景下需要根據索引的別名（ alias ）逐個設置，而 Data stream 則是拋棄了 alias 的限制，可以直接批量化設置相似名稱的索引，大大增加了 ILM 的使用範圍。

Data stream 的組成

數據流在 Elasticsearch 集群中由一個或多個隱藏的、自動生成的後備索引組成。

在實際的 Elasticsearch 操作中，數據流依靠索引模板來設定數據流實體的後備索引。

• 模板包含用於配置流的後備索引的映射和設置。
• 同一個索引模板可用於多個數據流。
• 不能刪除數據流正在使用的索引模板。

每個索引到數據流的文檔，必須包含一個 @timestamp 字段，映射為 date 或 date_nanos 字段類型。如果索引模板沒有為 @timestamp 字段指定映射，Elasticsearch 將 @timestamp 映射為帶有默認選項的日期字段。

Data stream 的讀請求主要如下圖，數據流自動將請求路由到其所有後備索引。

而對於寫請求，數據流則將該請求自動轉發給最新的後備索引。

對於寫請求，有兩點需要注意：

• 不能將新文檔添加到其他非最新後備索引，即使直接將請求發送到這些索引也不行。
• 不能對正在寫入的索引做 Clone/Close/Delete/Freeze/Shrink/Split 相關操作。

注：7.12版本可以 Close

Data stream 的特性

生成

每個 Data stream 的後備索引都有一個 generation 數，一個六位數，零填充的整數，從 000001 開始，用作該流的 rollover 的計數。

後備索引名主要依照以下格式：

.ds--

Generation 越大，後備索引包含的數據越新。 例如，web-server-logs 數據流最新的 generation 為 34。該流的最新後備索引名為 .ds-web-server-logs-000034。

注意：某些操作（例如 shrink 或 restore ）可以更改後備索引的名稱。 這些名稱更改不會從其數據流中刪除後備索引。

Rollover

在 Data stream 的使用中，rollover 是必不可少的條件。

創建數據流時，Elasticsearch 會自動為該 Data stream 根據 template 創建一個後備索引。 該索引還充當流的第一個寫入索引。當滿足一定條件時， rollover 會創建一個新的後備索引，該後備索引將成為 Data stream 的新寫入索引。

當然 rollover 的條件設置主要依靠 ILM。 如果需要，你還可以手動將數據 rollover 。

追加

由於時序性數據的特徵，Data stream 的設計場景中，數據是隻追加的，極少需要修改刪除。如果實際需要修改刪除，則可以考慮以下操作：

• 對於數據流只能通過 update by query 或者 delete by query 操作，不能進行 update 或者 delete 文檔。
• 需要 delete 或者 update 文檔，則直接對後備索引操作。
• 需要經常刪除或者修改文檔的，請使用索引別名或者索引模板，不要對 Data stream 操作。

Data stream 的使用

創建索引生命週期管理策略 ILM

索引生命週期管理策略 ILM 的主要配置細節見索引週期管理一章，此處主要做 hot 和 delete 階段的設置，用於 rollover 的引用。

相關命令：

PUT /_ilm/policy/my-data-stream-policy
{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {
            "max_size": "25GB"
          }
        }
      },
      "delete": {
        "min_age": "30d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

Kibana 圖形界面： Stack Management -> Index Lifecycle Policies -> Create policy

注意：

1. rollover 設置中，文檔數和最大存在時間是相對敏感的配置參數，由於 Elasticsearch 並不是實時監控 ILM 的執行任務（默認十分鐘），最終結果並不一定完全一致。
2. ILM 任務判斷中，max_size 判斷的是主分片的大小，而不是整個索引的大小。
3. 新版本下，max_size 的判斷並不敏感，可能是因為索引的主分片 size 大小會被 merge 後收縮，需要有一定時間的觀察。如下圖。測試之下，200MB之下的 max_size 會失效。建議 max_size 設置參數不要太小。

創建索引模板

索引模板是後備索引設置，以及 mapping 的主要配置來源，此處不展開延伸。主要設置 Data stream 相關的部分。

相關命令：

PUT /_index_template/my-data-stream-template
{
  "index_patterns": [ "my-data-stream*" ],
  "data_stream": { },
  "priority": 200,
  "template": {
    "settings": {
      "index.lifecycle.name": "my-data-stream-policy"
    }
  }
}

注意：

1. 定義 data_stream 為一個空的 object ，這是必要的。
2. Template 中使用了上一步創建的 ILM 策略 my-data-stream-policy。

此外，還需要注意兩點：

1. Elasticsearch 有一些內置索引模板如 metric-- 和 logs-- ，默認優先級 priority 是 100。如果有重名使用，則可以調高優先級，防止被默認的覆蓋。
2. 索引模板默認將 @timestamp 字段設置為 date 屬性。

Kibana 界面:

Stack Management -> Index Management -> Index Templates -> Create template

創建 template，不要創建舊版索引，並打開數據流標籤

設置生命週期管理策略，其他設置此處省略，一直下一步至創建完成。

創建 Data stream

可以自動利用 template 的匹配模式新增文檔創建：

POST /my-data-stream/_doc/
{
  "@timestamp": "2020-12-06T11:04:05.000Z",
  "user": {
    "id": "vlb44hny"
  },
  "message": "Login attempt failed"
}

Response:

{
  "_index" : ".ds-my-data-stream-000001",
  "_type" : "_doc",
  "_id" : "8ZadZXkBkhA9X9yUbI17",
  "_version" : 1,
  "result" : "created",
  "_shards" : {
    "total" : 2,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 0,
  "_primary_term" : 1
}

也可以直接 PUT 創建一個空的 Data stream。

PUT /_data_stream/my-data-stream

刪除

刪除命令：

DELETE /_data_stream/my-data-stream

刪除數據流會將數據流的後備索引一起刪除。

使用 Data stream

此處對數據流的操作主要以命令為主，Kibana 界面支持較少。

新增數據

Data stream 在新增數據時是隻追加的模式，因此在固定 id 和 bulk 的模式下，op_type 是指定 create 的。

如下面命令：

POST my-data-stream/_create/1
{"@timestamp":"2020-12-07T11:06:07.000Z","test":1}

或者

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" }

如果並不指定，文檔的 id，則可以使用默認的 _doc ，如下：

POST my-data-stream/_doc/
{"@timestamp":"2020-12-07T11:06:07.000Z","test":1}

獲取 Data stream 狀態

使用 Data stream stats API 查看 Data stream 的狀態。

GET /_data_stream/my-data-stream/_stats?human=true

Response:

{
  "_shards" : {
    "total" : 4,
    "successful" : 2,
    "failed" : 0
  },
  "data_stream_count" : 1,
  "backing_indices" : 1,
  "total_store_size" : "5kb",
  "total_store_size_bytes" : 5151,
  "data_streams" : [
    {
      "data_stream" : "my-data-stream",
      "backing_indices" : 1,
      "store_size" : "5kb",
      "store_size_bytes" : 5151,
      "maximum_timestamp" : 1607252645000
    }
  ]
}

可見 my-data-stream 的大下和後備索引數量。

同時需要用 _ilm/explain 獲取 Data stream 後備索引所在的 ILM 策略狀態。

GET my-data-stream/_ilm/explain

Response:

{
  "indices" : {
    ".ds-my-data-stream-000001" : {
      "index" : ".ds-my-data-stream-000001",
      "managed" : true,
      "policy" : "my-data-stream-policy",
      "lifecycle_date_millis" : 1620907943375,
      "age" : "7.95s",
      "phase" : "hot",
      "phase_time_millis" : 1620907943567,
      "action" : "rollover",
      "action_time_millis" : 1620907943661,
      "step" : "check-rollover-ready",
      "step_time_millis" : 1620907943661,
      "phase_execution" : {
        "policy" : "my-data-stream-policy",
        "phase_definition" : {
          "min_age" : "0ms",
          "actions" : {
            "rollover" : {
              "max_size" : "25gb"
            }
          }
        },
        "version" : 1,
        "modified_date_in_millis" : 1620907939978
      }
    }
  }
}

上圖可見，這個數據的 000001 索引主要處於 hot 階段，策略名稱是 logs 等信息。具體參數可見於 ILM 的相關定義。

手動 rollover Data stream

使用 rollover API，手動 rollover Data stream。

POST my-data-stream/_rollover

Response:

{
  "acknowledged" : true,
  "shards_acknowledged" : true,
  "old_index" : ".ds-my-data-stream-000001",
  "new_index" : ".ds-my-data-stream-000002",
  "rolled_over" : true,
  "dry_run" : false,
  "conditions" : { }
}

再 GET 相關 Data stream 狀態，後備索引增加。

GET /_data_stream/my-data-stream/

Response:

{
  "data_streams" : [
    {
      "name" : "my-data-stream",
      "timestamp_field" : {
        "name" : "@timestamp"
      },
      "indices" : [
        {
          "index_name" : ".ds-my-data-stream-000001",
          "index_uuid" : "AJBi0g3fRyG8-1tiH2UD2Q"
        },
        {
          "index_name" : ".ds-my-data-stream-000002",
          "index_uuid" : "AgOLGMSBSYWb4X-ID8uwtg"
        }
      ],
      "generation" : 2,
      "status" : "GREEN",
      "template" : "my-data-stream-template",
      "ilm_policy" : "my-data-stream-policy",
      "hidden" : false
    }
  ]
}

Reindex Data stream

使用 reindex API 去複製數據到一個 Data stream。由於 Data stream 的只追加特性，在 op_type 中要選擇為 create 。

POST /_reindex
{
  "source": {
    "index": "test"
  },
  "dest": {
    "index": "my-data-stream",
    "op_type": "create"
  }
}

Response:

{
  "took" : 80,
  "timed_out" : false,
  "total" : 1,
  "updated" : 0,
  "created" : 1,
  "deleted" : 0,
  "batches" : 1,
  "version_conflicts" : 0,
  "noops" : 0,
  "retries" : {
    "bulk" : 0,
    "search" : 0
  },
  "throttled_millis" : 0,
  "requests_per_second" : -1.0,
  "throttled_until_millis" : 0,
  "failures" : [ ]
}

Delete/Update by query

針對 Data stream 只能 delete/update by query 。

相關命令：

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"
    }
  }
}

POST /my-data-stream/_delete_by_query
{
  "query": {
    "match": {
      "user.id": "vlb44hny"
    }
  }
}

Delete update 後備索引數據

在後備索引刪除或者修改，需要注意下面三個要素:

• 文檔 id。
• 文檔所在的後備索引。
• 如果是修改文檔，則需要其 _seq_no 和 _primary_term 兩個參數。

主要操作如下：

先獲取文檔所需的要素信息，設置 seq_no_primary_term 為 true。

GET /my-data-stream/_search
{
  "seq_no_primary_term": true,
  "query": {
    "match": {
      "message": "Login attempt failed"
    }
  }
}

獲得結果：

{
  "took" : 621,
  "timed_out" : false,
  "_shards" : {
    "total" : 2,
    "successful" : 2,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 0.8630463,
    "hits" : [
      {
        "_index" : ".ds-my-data-stream-000001",
        "_type" : "_doc",
        "_id" : "9ZakZXkBkhA9X9yUZo2P",
        "_seq_no" : 0,
        "_primary_term" : 1,
        "_score" : 0.8630463,
        "_source" : {
          "@timestamp" : "2020-12-06T11:04:05.000Z",
          "user" : {
            "id" : "vlb44hny"
          },
          "message" : "Login attempt failed"
        }
      }
    ]
  }
}

然後修改命令：

PUT /.ds-my-data-stream-000001/_doc/9ZakZXkBkhA9X9yUZo2P?if_seq_no=0&if_primary_term=1
{
  "@timestamp": "2020-12-07T11:06:07.000Z",
  "test": 4
}

Response:

{
  "_index" : ".ds-my-data-stream-000001",
  "_type" : "_doc",
  "_id" : "9ZakZXkBkhA9X9yUZo2P",
  "_version" : 2,
  "result" : "updated",
  "_shards" : {
    "total" : 2,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 1,
  "_primary_term" : 1
}

或者刪除命令：

DELETE  /.ds-logs-1-1-000002/_doc/3

關於修改 Data stream 的 mapping 和 setting

Data stream 的 setting 和 mapping 修改主要還是基於 Elasticsearch 默認的修改規則。總結一下，主要有以下幾點：

• 新增字段不影響。
• 已存在的配置不可更改。
• 修改的 template 只能應用於未來新增的索引。

因此，如果需要修改不可更改的配置，可以考慮 reindex 或者修改 template 後手工 Rollover Data stream。

關於 Data tiers

Data tiers 也稱數據層，是一個在 7.10 版本的一個新概念。

Data tiers 主要的一個特點是將節點角色（ node roles ）與索引生命週期所需要的節點屬性（ attribute ）結合，直接可以在制定 Elasticsearch 節點角色時配置，不需要再去設置 attribute 。Data tiers 的概念也是對時序性數據分層管理的優化配置。

Data tiers 的數據節點默認是都配置的，即 data_content/data_hot/data_warm/data_cold（ chsw ）都具備。

Tiers 的定義

• Content tier

Content tier 節點存儲的數據，往往定義為與時序性數據相反的常態化數據，比如商品種類這種隨著時間推移保持相對不變。這種數據並不能根據冷熱數據性質分層存儲。

此類數據有以下特點：

- Content tier 節點通常需要較高的計算性能，要求處理能力比 IO 吞吐能力高，需要處理複雜的搜索和聚合並快速返回結果。

- 對數據內容的獲取，即文檔內容本身獲取比時序性數據要少。

- 這類數據索引需要配置為一個或多個副本。

• Hot tier

Hot tier，熱層是時間序列數據的 Elasticsearch 入口點，最新存儲的時間序列數據。Hot tier 的數據也是會被查詢最多的數據。因此熱層中的節點在讀取和寫入時都需要快速，這需要更多的硬件資源和更快的存儲（ SSD）。屬於數據流 （ Data stream ）的新索引會自動分配給熱層。

• Warm tier

即溫層，一旦查詢時間序列數據的頻率低於 hot tier 中最近索引的數據，便可以將其移至 warm tier 。 warm tier 通常保存最近幾周的數據。 仍然允許進行更新，但可能很少。通常，warm tier 中的節點不需要像 hot tier 中的節點一樣快。

• Cold tier

冷層的數據一般查詢頻率非常低，且不會被更新。 但是 cold tier 仍然是響應查詢層。 隨著數據過渡到 cold tier，可以對其進行壓縮和去副本。Cold tier節點的機器配置可以相對較低。

tier_preference

index.routing.allocation.include._tier_preference 是 Data tiers 的主要配置方式，在分片數據的時候使用 tier_preference 指定數據節點的分配。

tier_preference 的設置會有三種情況：

1. 創建正常索引時，默認情況下，Elasticsearch 將 index.routing.allocation.include._tier_preference 設置為 data_content ，以將索引分片自動分配給內容層。
2. 創建數據流時，Elasticsearch 會將後備索引的 index.routing.allocation.include._tier_preference 設置為 data_hot，以自動將索引分片分配給熱層。
3. 顯式設置 index.routing.allocation.include._tier_preference，選擇索引需要的數據節點。 如果將層首選項設置為 null，則 Elasticsearch 在分配期間將忽略數據層角色，依照其它參數分配。

相關的圖形和命令配置如下：

上圖時在索引生命週期管理中選擇 Data tiers節點。

PUT _index_template/template_demo
{
  "index_patterns": ["demo-*"],
  "data_stream": {},
  "priority": 200,
  "template": {
    "settings": {
      "number_of_shards": 2,
      "index.lifecycle.name": "demo",
      "index.routing.allocation.include._tier_preference": "data_hot"
    }
  }
}

上面命令中設置索引模板匹配 demo-* 的索引的分配策略為 "index.routing.allocation.include._tier_preference":"data_hot"