Elasticsearch的ETL利器——Ingest節點

一、問題引出
來自星球同窗的提問：

「Ingest node什麼場景會遇到它？ 一直沒搜到它是在什麼場景工做的？」

的確咱們比較關心集羣的節點角色的劃分。包括：

集羣應該幾個節點？

幾個節點用於數據存儲？

要不要獨立Master節點、協調節點？

可是Ingest node的場景用的比較少。

二、集羣節點角色劃分梳理
以前的文章：刨根問底 | Elasticsearch 5.X集羣多節點角色配置深刻詳解有過解讀。本文再參考7.1版本官方文檔總結一下：

2.1 主節點
主節點負責集羣相關的操做，例如建立或刪除索引，跟蹤哪些節點是集羣的一部分，以及決定將哪些分片分配給哪些節點。

擁有穩定的主節點是衡量集羣健康的重要標誌。

注意：

一、因爲索引和搜索數據都是CPU、內存、IO密集型的，可能會對數據節點的資源形成較大壓力。 所以，在較大規模的集羣裏，最好要設置單獨的僅主節點角色。（這點PB級集羣調優時重點關注）

二、不要將主節點同時充當協調節點的角色，由於：對於穩定的集羣來講，主節點的角色功能越單一越好。

2.2 數據節點
數據節點：保存包含索引文檔的分片數據，執行CRUD、搜索、聚合相關的操做。屬於：內存、CPU、IO密集型，對硬件資源要求高。

2.3 協調節點
搜索請求在兩個階段中執行（query 和 fetch），這兩個階段由接收客戶端請求的節點 - 協調節點協調。

在請求階段，協調節點將請求轉發到保存數據的數據節點。 每一個數據節點在本地執行請求並將其結果返回給協調節點。

在收集fetch階段，協調節點將每一個數據節點的結果聚集爲單個全局結果集。

2.4 Ingest節點
ingest 節點能夠看做是數據前置處理轉換的節點，支持 pipeline管道 設置，可使用 ingest 對數據進行過濾、轉換等操做，相似於 logstash 中 filter 的做用，功能至關強大。

我把Ingest節點的功能抽象爲：大數據處理環節的「ETL」——抽取、轉換、加載。

前Elastic中國架構師吳斌的文章中對Ingest節點的評價很高，他指出

「2018這一年來拜訪了不少用戶，其中有至關一部分在數據攝取時遇到包括性能在內的各類各樣的問題，那麼大多數在咱們作了ingest節點的調整後獲得了很好的解決。Ingest節點不是萬能的，可是使用起來簡單，並且拋開後面數據節點來看性能提高趨於線性。因此我一直本着能用ingest節點解決的問題，毫不麻煩其餘組件的大致原則」。

三、Ingest 節點能解決什麼問題？
上面的Ingest節點介紹太官方，看不大懂怎麼辦？來個實戰場景例子吧。

思考問題1：線上寫入數據改字段需求

如何在數據寫入階段修改字段名（不是修改字段值）？

思考問題2：線上業務數據添加特定字段需求

如何在批量寫入數據的時候，每條document插入實時時間戳？

這時，腦海裏開始對已有的知識點進行搜索。
針對思考問題1：字段值的修改無非：update,updatebyquery？可是字段名呢？貌似沒有相關接口或實現。

針對思考問題2：插入的時候，業務層面處理，讀取當前時間並寫入貌似能夠，有沒有不動業務層面的字段的方法呢？

答案是有的，這就是Ingest節點的妙處。

四、Ingest實踐一把
針對問題1：

PUT _ingest/pipeline/rename_hostname

{

"processors": [

{

    "field": "hostname",

    "target_field": "host",

    "ignore_missing": true

  }

}

]

}

PUT server

POST server/values/?pipeline=rename_hostname

{

"hostname": "myserver"

}

如上，藉助Ingest節點的 rename_hostname管道的預處理功能，實現了字段名稱的變動：由hostname改爲host。

針對問題2：6.5版本ES驗證ok以下。

PUT _ingest/pipeline/indexed_at

{

"description": "Adds indexed_at timestamp to documents",

"processors": [

{

  "set": {

    "field": "_source.indexed_at",

    "value": "{{_ingest.timestamp}}"

  }

}

]

}

PUT ms-test

{

"settings": {

"index.default_pipeline": "indexed_at"

}

}

POST ms-test/_doc/1

{"title":"just testing"}

如上，經過indexedat管道的set處理器與ms-test的索引層面關聯操做，
ms-test索引每插入一篇document，都會自動添加一個字段indexat=最新時間戳。

五、Ingest節點基本概念
在實際文檔索引起生以前，使用Ingest節點預處理文檔。Ingest節點攔截批量和索引請求，它應用轉換，而後將文檔傳遞迴索引或Bulk API。

強調一下： Ingest節點處理時機——在數據被索引以前，經過預約義好的處理管道對數據進行預處理。

默認狀況下，全部節點都啓用Ingest，所以任何節點均可以處理Ingest任務。咱們也能夠建立專用的Ingest節點。

要禁用節點的Ingest功能，須要在elasticsearch.yml 設置以下：

node.ingest：false

這裏就涉及幾個知識點：

一、預處理 pre-process

要在數據索引化(indexing)以前預處理文檔。

二、管道 pipeline

每一個預處理過程能夠指定包含一個或多個處理器的管道。

管道的實際組成：

{

"description" : "...",

"processors" : [ ... ]

}

description：管道功能描述。
processors：注意是數組，能夠指定1個或多個處理器。

三、處理器 processors

每一個處理器以某種特定方式轉換文檔。

例如，管道可能有一個從文檔中刪除字段的處理器，而後是另外一個重命名字段的處理器。 這樣，再反過來看第4部分就很好理解了。

六、Ingest API
Ingest API共分爲4種操做，分別對應：

PUT（新增）、

GET（獲取）、

DELETE（刪除）、

Simulate （仿真模擬）。

模擬管道AP Simulate 針對請求正文中提供的文檔集執行特定管道。

除此以外，高階操做包括：

一、支持複雜條件的Nested類型的操做；

二、限定條件的管道操做；

三、限定條件的正則操做等。

詳細內容，參見官網便可。

常見的處理器有以下28種，舉例：

append處理器：添加1個或1組字段值；

convert處理器：支持類型轉換。

建議：不必都過一遍，根據業務需求，反查文檔便可。

七、Ingest節點和Logstash Filter 啥區別？
業務選型中，確定會問到這個問題。

區別一：支持的數據源不一樣。
Logstash：大量的輸入和輸出插件（好比：kafka，redis等）可供使用，還可用來支持一系列不一樣的架構。
Ingest節點：不能從外部來源（例如消息隊列或數據庫）提取數據，必須批量bulk或索引index請求將數據推送到 Elasticsearch.

區別二：應對數據激增的能力不一樣。
Logstash：Logstash 可在本地對數據進行緩衝以應對採集驟升狀況。如前所述，Logstash 同時還支持與大量不一樣的消息隊列類型進行集成。
Ingest節點：極限狀況下會出現：在長時間沒法聯繫上 Elasticsearch 或者 Elasticsearch 沒法接受數據的狀況下，均有可能會丟失數據。

區別三：處理能力不一樣。
Logstash：支持的插件和功能點較Ingest節點多不少。
Ingest節點：支持28類處理器操做。Ingest節點管道只能在單一事件的上下文中運行。Ingest一般不能調用其餘系統或者從磁盤中讀取數據。

區別四：排他式功能支持不一樣。
Ingest節點：支持採集附件處理器插件，此插件可用來處理和索引常見格式（例如 PPT、XLS 和 PDF）的附件。
Logstash：不支持如上文件附件類型。

選型小結：

一、兩種方式各有利弊，建議小數據規模，建議使用Ingest節點。緣由：架構模型簡單，不須要額外的硬件設備支撐。

二、數據規模大以後，除了建議獨立Ingest節點，同時建議架構中使用Logstash結合消息隊列如Kafka的架構選型。

三、將Logstash和Ingest節點結合，也是架構選型參考方案之一。

八、小結
Ingest是5.X版本就有的特性，不算是新知識點。

實踐很重要！當咱們對不熟悉的概念學習的時候，能夠先查一下別人的應用場景，大體理解一下，再動手跟着官方文檔敲一遍、理解一遍，加深認知。

基於Ingest實現的PDF文檔預處理和索引，甚至基於Ingest自定義插件開發能夠實現更多複雜的功能，你均可以嘗試一下！

參考：

https://0x7.me/oLX3m

https://elasticsearch.cn/article/6221

https://discuss.elastic.co/t/etl-tool-for-elasticsearch/113803/9

https://github.com/bnafziger/elasticsearch-ingest-opennlp