elasticsearch的ingest node讲解以及Pipeline和grok分析

古城微笑少年丶 2022-12-28 00:42 167阅读 0赞

> 参考资料  
> https://blog.csdn.net/UbuntuTouch/article/details/99702199

# 需求 #

最近在做日志收集到es时遇到些问题。我不想用logstash做日志的切分，毕竟很大，想用FileBeat直接推数据到es中。这就涉及到一个问题，这步切分数据在哪做。这个可以使用es的ingest node来做。

# 解决 #

### ingest node ###

es在5.0X以后推出了ingest node的功能，可以对数据进行预处理。我们可以指定需要的Pipeline，然后ingest node 将会帮我们按照规定的 processor 顺序来执行对数据的操作和处理。

### 定义pipeline ###

要在索引之前预处理文档，我们必须定义pipeline。pipeline定义了一系列processor。每个processor以某种方式转换文档。每个processor按照在 pipeline 中定义的顺序执行。 pipeline由两个主要字段组成description和processor列表。

pipeline 是群集级存储而被保存在每个节点的内存中，并且 pipeline 始终在 ingest node中运行，因此最好在群集中保留需要的 pipeline，而删除那些不需要的 pipeline。

Pipeline 以 cluster 状态存储，并且立即传播到所有 ingest node。 当 ingest node 接收到新 pipeline 时，它们将以内存 pipeline 表示形式更新其节点，并且 pipeline 更改将立即生效。

##### 命令相关 #####

注册或者更新pipeline（根据对应的json注册test-pipeline）

> curl -H “Content-Type: application/json” -XPUT ‘http://10.130.7.207:9200/\_ingest/pipeline/test-pipeline’ -d@/home/hadoop/zgh/log/pipeline.json

删除pipeline（下面是删除test-pipeline）

> curl -XDELETE ‘10.130.7.207:9200/\_ingest/pipeline/test-pipeline?pretty’

##### pipeline的json格式 #####

格式如下

{
         "description" : "...",
         "processors" : [ ... ]
    }

### processor ###

我在另外一篇文章也有讲怎么将FileBeat的数据直接传到es，然后做切分。这里只是讲下processor。

ingest node内置了很多processor，grok就是属于processor的一种。processor还有有很多，参考以下网址

> 内置的processor  
> https://www.elastic.co/guide/en/elasticsearch/reference/current/ingest-processors.html

我们这里2中常见的grok和drop

##### grok #####

grok 适用于正则匹配，用来匹配分割文本并映射到关键字的工具,在grok调试中，由于采用是整个字段匹配，如果一处无法匹配，就会导致整个匹配失败  
我们在写grok的时候，需要了解它的自带的类型，以及正则相关的知识，然后进行测试，下面是相关的链接

> grok的pattern  
> https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Apx-GrokPatterns/GrokPatterns\_title.html

> 测试grok的工具  
> http://grokdebug.herokuapp.com

这里是些常用的

> GREEDYDATA 匹配出\\n外的所有值  
> LOGLEVEL 匹配日志级别  
> TIMESTAMP\_ISO8601 匹配时间 毫秒级别

遇到的问题： 我在匹配正则的时候，想匹配全部使用GREEDYDATA 发现失败。因为GREEDYDATA 不匹配\\n， 后来使用(.|\\n)\*这种形式发现是可以的。

"grok": {
      "field": "message",
      "patterns": ["%{TIMESTAMP_ISO8601:client}    %{GREEDYDATA:method}    %{LOGLEVEL:request}    %{GREEDYDATA:demo}    %{ALL_CODE:messageaaa}" ],
      "pattern_definitions" : {
       	"ALL_CODE" : "(.|\n)*"
      }
    }

##### remove #####

remove这里用于删除字段。我这边的需求用到它是因为使用grok切分完，原有字段就将其删除了。  
格式为,这个看官网就行，一大片

{
             "remove": {
                 "field" : "message"
             }
      }