如何在Elasticsearch中使用pipeline API来对事件进行处理

╰半夏微凉° 2023-06-14 15:53 99阅读 0赞

### 1. 前言 ###

在`Elasticsearch 5.0`之前，如果我们想在将文档索引到Elasticsearch之前预处理文档，那么唯一的方法是使用Logstash或以编程方式/手动预处理它们，然后将它们索引到Elasticsearch。 Elasticsearch缺乏预处理/转换文档的能力，它只是按原样索引文档。 但是，在Elasticsearch 5.x之后引入一个名为`ingest node`的功能，为Elasticsearch本身的文档预处理和丰富之前提供了一个轻量级的解决方案。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70]

如果使用默认配置实现Elasticsearch节点，则默认情况下将启用`master`，`data`和`ingest`（即，它将充当主节点，数据节点和提取节点）。 要在节点上禁用`ingest`，请在`elasticsearch.yml`文件中配置以下设置：

node.ingest: false

`ingest`节点可用于在对文档执行实际索引之前预处理文档。 此预处理通过截取批量和索引请求的摄取节点执行，它将转换应用于数据，然后将文档传递回索引或批量API。 随着新的摄取功能的发布，Elasticsearch已经取出了Logstash的过滤器部分，以便我们可以在Elasticsearch中处理原始日志和丰富。

要在索引之前预处理文档，我们必须定义`pipeline`（其中包含称为处理器的步骤序列，用于转换传入文档）。 要使用`pipeline`，我们只需在索引或批量请求上指定`pipeline`参数，以告诉摄取节点使用哪个`pipeline`：

POST my_index/my_type?pipeline=my_pipeline_id 
    { 
         "key": "value"
    }

### 2. 定义一个`pipeline` ###

`pipeline`定义了一系列处理器。 每个处理器以某种方式转换文档。 每个处理器按照在`pipeline`中定义的顺序执行。 `pipeline`由两个主要字段组成：`description`和`processor`列表。  
`description`参数是一个非必需字段，用于存储一些描述/管道的用法; 使用`processor`参数，可以列出处理器以转换文档。`processor`的典型结构如下所示：

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

`ingest`节点有大约`20个`内置`processor`，包括`gsub`，`grok`，`转换`，`删除`，`重命名`等。 这些可以在构建管道时使用。 除了内置`processor`外，还可以使用`摄取附件`（如`ingest attachment`，`ingetst geo-ip`和`ingest user-agent`）等摄取插件，并可在构建`pipeline`时使用。 这些插件在默认情况下不可用，可以像任何其他Elasticsearch插件一样进行安装。

### 3. `Ingest APIs` ###

`ingest`节点提供一组称为`inget API`的API，可用于`定义`，`模拟`，`删除`或`查找`有关`pipeline`的信息。 摄取API端点是`_ingest`。

#### 3.1 `Put pipeline API` ####

此API用于定义新`pipeline`。 此API还用于添加新`pipeline`或更新现有`pipeline`。  
我们来看一个例子吧。 如下面的代码所示，我们定义了一个名为`firstpipeline`的新`pipeline`，它将消息字段中的值转换为大写：

PUT _ingest/pipeline/firstpipeline
    { 
      "description": "uppsercase the incoming vlaue in the message filed",
      "processors": [
        { 
          "uppercase": { 
            "field": "message"
          }
        }
      ]
    }

我们在Kibana中执行上面的命令，我们可以看到成功返回：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 1]  
让我们来测试一下我们新建立的`pipeline`：

PUT myindex/_doc/1?pipeline=firstpipeline
    { 
      "name": "pipeline",
      "message": "this is so cool!"
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 2]

我们看见我们的文档印被成功创建并存于一个叫做`myindex`的`index`里。下面我们来查看一下，我们刚才定义的`pipeline`是否已经起作用了：

GET myindex/_doc/1

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 3]

我们可以看到我们的message已经都变成大写的了。

创建管道时，可以定义多个处理器，执行顺序取决于定义中定义的顺序。 让我们看一个这样的例子。 如下面的代码所示，我们创建了一个名为`secondpipeline`的新管道，它转换“message”字段中存在的大写值，并将“message”字段重命名为“data”。 它创建一个名为“label”的新字段，其值为testlabel：

PUT _ingest/pipeline/secondpipeline
    { 
      "description": "uppercase the incoming value in the message field",
      "processors": [
        { 
          "uppercase": { 
            "field": "message",
            "ignore_failure": true
          }
        },
        { 
          "rename": { 
            "field": "message",
            "target_field": "data",
            "ignore_failure": true
          }
        },
        { 
          "set": { 
            "field": "label",
            "value": "testlabel",
            "override": false
          }
        }
      ]
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 4]

我们的第二个被叫做`secondpipeline`的也已经创建好了。接下来，让我们来利用这个`pipeline`来对我们的文档进行处理。我们在Kibnana中输入：

PUT myindex/_doc/1?pipeline=secondpipeline
    { 
      "name": "pipeline",
      "message": "this is so cool!"
    }

然后，我们使用如下的命令来查询我们刚才输入的文档：

GET myindex/_doc/1

显示的结果如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 5]  
通过上面的例子，我们可以看到我们之前的message项不见了，取而代之的是data，同时它里面的字符都变成大写的了。另外，它也新增加了一个叫做label的项，并且它的值被设置为testlabel。

> **提示：**  
> 如果缺少处理器中使用的字段，则处理器将抛出异常，并且不会对文档编制索引。 为了防止处理器抛出异常，我们可以利用  
> `“ignore_failure”：true`参数。

#### 3.2 `获取 pipeline API` ####

此API用于检索现有`pipeline`的定义。 使用此API，可以找到单个pipeline定义的详细信息或查找所有pipeline的定义。

查找所有pipeline定义的命令是：

GET _ingest/pipeline

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 6]

要查找现有`pipeline`的定义，请将管道ID传递给get管道.api。 以下是查找名为secondpipeline的pipeline定义的示例：

GET _ingest/pipeline/secondpipeline

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 7]

#### 3.3 `删除 pipeline API` ####

删除管道API`按ID`或`通配符匹配`删除pipeline。 以下是删除名为firstpipeline的pipeline的示例：

DELETE _ingest/pipeline/firstpipeline

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 8]

这样firstpipeline就被删除了。

#### 3.4 `模拟 pipeline API` ####

此`pipeline API`可用于根据请求正文中提供的文档集模拟pipeline的执行。 可以指定要对提供的文档执行的现有pipeline，或者在请求的主体中提供pipeline定义。 要模拟ingest pipeline，请将“`_simulate`”端点添加到`pipeline API`。

以下是模拟现有`pipeline`的示例：

POST _ingest/pipeline/secondpipeline/_simulate
    { 
      "docs": [
        { 
          "_source": { 
            "name": "pipeline",
            "message": "this is so cool!"
          }
        },
        { 
          "_source": { 
            "name": "nice",
            "message": "this is nice!"
          }
        }
      ]
    }

执行的结果是：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 9]

我们可以在右边看出来执行pipeline所显示的结果。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20190818111726970.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20190818110144341.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20190818111237325.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20190818111310738.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20190818113542724.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20190818113945606.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20190818141117111.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20190818141444718.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20190818141811645.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo_size_16_color_FFFFFF_t_70 9]: https://img-blog.csdnimg.cn/20190818142525654.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo,size_16,color_FFFFFF,t_70