【最佳實踐】如何使用 Elasticsearch ingest 節點來豐富日誌和指標

從歷史上看，數據豐富功能僅在 Logstash 中可用，但由於 Elasticsearch 7.5.0 中引入了 enrich 處理器，因此可以直接在 Elasticsearch 中進行豐富，而無需配置單獨的服務/系統。如果你想知道在 Logstash 中是如何實現，那麼請參閱我之前的文章 “Logstash：運用 jdbc_streaming 來豐富我們的數據”。

由於通常用於豐富的主數據通常是在 CSV 文件中創建的，因此在此博客中，我們將逐步說明如何使用 CSV 文件中的數據將在攝取節點上運行的 enrich 處理器用於豐富數據。

樣本 CSV 數據

我們可以使用阿里雲Elasticsearch中的 Kibana ，或通過 ECS 自建 ELK，導入以下 CSV 格式的示例主數據，然後在將文檔吸收到 Elasticsearch 中時用於豐富文檔。對於本博客中給出的示例，我們將主數據存儲在一個名為 test.csv 的文件中。此數據代表組織清單中的設備。

test.csv

"Device ID","Device Location","Device Owner","Device Type"
"device1","London","Engineering","Computer"
"device2","Toronto","Consulting","Mouse"
"device3","Winnipeg","Sales","Computer"
"device4","Barcelona","Engineering","Phone"
"device5","Toronto","Consulting","Computer"
"device6","London","Consulting","Computer"

請注意，CSV 數據不應包含任何其他空格，因為當前版本的 Data Visualizer 需要精確格式化數據。在此 github 問題中對此進行了記錄。

將 CSV 數據導入 Elasticsearch

我們可以直接使用 Kibana 來把數據導入。打開 Kibana:

點擊上面的 Import a CSV, NDJSON, or log file 鏈接：、

點擊 Select or drag and drop a file, 然後選擇剛才我們創建的 test.csv 文件：

點擊 Import 按鈕：

我們把導入的這個 index 的名字叫做 master_data_from_csv。點擊 Import 按鈕：

這樣就完成了我們的 master_data_from_csv 的索引創建。我們可以在上面屏幕的下方的四個選項中任意選擇一個來查看導入的數據。

利用我們的主數據豐富文檔

在本節中，我們演示如何使用 enrich Processor 將主數據合併到輸入數據流中的文檔中。關於 enrich processor，我之前有另外一篇文章有詳述“Elasticsearch：enrich processor （7.5發行版新功能）”。

第一步是創建一個豐富的策略，該策略定義我們將使用哪個字段將主數據與輸入數據流中的文檔進行匹配。下面提供了適用於我們的數據的示例策略：

PUT /_enrich/policy/enrich-devices-policy
{
  "match": {
    "indices": "master_data_from_csv",
    "match_field": "Device ID",
    "enrich_fields": [
      "Device Location",
      "Device Owner",
      "Device Type"
    ]
  }
}

運行上面的策略。然後，我們使用 execute enrich policy API 為該策略創建 enrich 索引：

PUT /_enrich/policy/enrich-devices-policy/_execute

接下來，我們創建一個使用我們的豐富策略的 ingest pipeline。

PUT /_ingest/pipeline/device_lookup
{
  "description": "Enrich device information",
  "processors": [
    {
      "enrich": {
        "policy_name": "enrich-devices-policy",
        "field": "device_id",
        "target_field": "my_enriched_data",
        "max_matches": "1"
      }
    }
  ]
}

我們插入一個文檔，並讓它使用上面定義的 ingest pipeline，如下所示：

PUT /device_index/_doc/1?pipeline=device_lookup
{
  "device_id": "device1",
  "other_field": "some value"
}

我們可以使用 GET API 查看導入的文檔，如下所示：

GET device_index/_doc/1

{
  "_index" : "device_index",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 1,
  "_seq_no" : 0,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "my_enriched_data" : {
      "Device Location" : "London",
      "Device Owner" : "Engineering",
      "Device ID" : "device1",
      "Device Type" : "Computer"
    },
    "device_id" : "device1",
    "other_field" : "some value"
  }
}

在上面，我們可以看出來在返回的文檔信息中，有多一個叫做 my_enriched_data 的字段。它包含了 Device Location, Device Owner, Device ID 及 Device Type。這些信息都來自於我們之前導入的 test.csv 文檔信息。enrich processor 通過關聯 device_id 為 device1 而從 master_data_from_csv 索引中獲得這些信息。也就是說我們的數據變得更多了，這也就是我們之前說的豐富了。

在索引設置中指定pipeline

在上面，我們通過導入時指定的 pipeline 來進行調用 enrich processor，但是在實際的應用場景中，我們更加傾向於把這個配置與 index 的設置中，而不是在請求的url中來指定特定的 pipeline。我們可以通過在 index 的配置中添加 index.default_pipeline 來完成。

PUT device_index/_settings
{
  "index.default_pipeline": "device_lookup"
}

現在，發送到 device_index 的所有文檔都將通過 device_lookup 管道，而無需在 URL 中使用 pipeline=device_lookup。我們可以使用下面的 PUT 命令來驗證它是否正常工作。

PUT /device_index/_doc/2
{
  "device_id": "device2",
  "other_field": "some value"
}

執行以下命令以查看我們剛剛提取的文檔：

GET device_index/_doc/2

那麼你可以看到如下所示的文檔：

{
  "_index" : "device_index",
  "_type" : "_doc",
  "_id" : "2",
  "_version" : 1,
  "_seq_no" : 1,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "my_enriched_data" : {
      "Device Location" : "Toronto",
      "Device Owner" : "Consulting",
      "Device ID" : "device2",
      "Device Type" : "Mouse"
    },
    "device_id" : "device2",
    "other_field" : "some value"
  }
}