Elasticsearch使用Ingest Pipeline进行数据预处理

矫情吗；* 2023-06-30 08:42 50阅读 0赞

> 本文基于Elasticsearch7.x

Elasticsearch可以使用自身的Ingest Pipeline功能进行数据预处理, 无须借助Logstash.

### Ingest Pipeline介绍 ###

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

Ingest Pipeline 就是在文档写入Data Node之前进行一系列的数据预处理, 进行数据预处理的就是processor, 一组处理器构成了Pipeline. 所有的预处理都在Ingest Node上执行, 默认情况下所有节点都是Ingest Node.

**常用的processor**

*  split processor  
    字符串切分成数组
 *  join processor  
    数组转化成字符串
 *  gsub processor  
    字符串替换
 *  set processor  
    创建或替换一个字段.
 *  remove processor  
    移除一个字段
 *  rename processor  
    重命名一个字段
 *  lowercase processor  
    字符串小写化
 *  upcase processor  
    字符串大写化
 *  script processor  
    使用painless脚本进行复杂的处理
 *  …

**语法**

(1) 创建pipeline

description是对pipeline的描述, processors定义了一组处理器.

PUT /_ingest/pipeline/my_pipeline_id
    {
      "description": "to split blog tags",
      "processors": [
        {
          "split": {
            "field": "tags",
            "separator": ","
          }
        }
      ]
    }

(2) 查询

GET /_ingest/pipeline/my_pipeline_id

(3) 删除

DELETE /_ingest/pipeline/my_pipeline_id

**实例**

(1) 创建pipeline

PUT /_ingest/pipeline/my_pipeline_id
    {
      "description": "to split blog tags",
      "processors": [
        {
          "split": {
            "field": "tags",
            "separator": ","
          }
        },
        {
          "set": {
            "field": "views",
            "value": "0"
          }
        }
      ]
    }

创建一个pipeline, 它的作用是将tags字段按逗号切分成数组, 创建或替换一个views字段, 值为0.

(2) 新增文档

新增一个文档, 指定pipeline为my\_pipeline\_id.

PUT /blogs/_doc/1?pipeline=my_pipeline_id
    {
      "title": "Introducing big data......",
      "tags": "hadoop,elasticsearch,spark",
      "content": "You konw, for big data"
    }

(3) 查看文档

GET /blogs/_doc/1

结果:

{
      "_index" : "blogs",
      "_type" : "_doc",
      "_id" : "1",
      "_version" : 1,
      "_seq_no" : 0,
      "_primary_term" : 1,
      "found" : true,
      "_source" : {
        "title" : "Introducing big data......",
        "content" : "You konw, for big data",
        "views" : "0",
        "tags" : [
          "hadoop",
          "elasticsearch",
          "spark"
        ]
      }
    }

可以看到tags字符串被切分成数组了, 同时新增了一个view字段.

**使用Painless脚本定义Processor**

Elasticsearch内置了很多processor, 可以进行一些简单的数据预处理, 但如果我们想进行复杂的数据预处理, 就需要使用painless脚本来自定义processor.

(1) 创建pipeline

我们使用painless脚本来自定义一个processor, 如果存在一个content的字段, 则新增一个content\_length, 值为content字段值的长度.

PUT /_ingest/pipeline/my_pipeline_id
    {
      "description": "to split blog tags",
      "processors": [
        {
          "split": {
            "field": "tags",
            "separator": ","
          }
        },
        {
          "script": {
            "source": """
              if(ctx.containsKey("content")){
                ctx.content_length = ctx.content.length();
              }else{
                ctx.content_length = 0;
              }
            """
          }
        }
      ]
    }

(2) 新增文档

PUT /blogs/_doc/1?pipeline=my_pipeline_id
    {
      "title": "Introducing big data......",
      "tags": "hadoop,elasticsearch,spark",
      "content": "You konw, for big data"
    }
    
    PUT /blogs/_doc/2?pipeline=my_pipeline_id
    {
      "title":"Introducing cloud computering",
      "tags":"openstack,k8s",
      "content":"You konw, for cloud"
    }

(3) 查看文档

GET /blogs/_search

结果:

"hits" : [
      {
        "_index" : "blogs",
        "_type" : "_doc",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "title" : "Introducing big data......",
          "content" : "You konw, for big data",
          "content_length" : 22,
          "tags" : [
            "hadoop",
            "elasticsearch",
            "spark"
          ]
        }
      },
      {
        "_index" : "blogs",
        "_type" : "_doc",
        "_id" : "2",
        "_score" : 1.0,
        "_source" : {
          "title" : "Introducing cloud computering",
          "content" : "You konw, for cloud",
          "content_length" : 19,
          "tags" : [
            "openstack",
            "k8s"
          ]
        }
      }
    ]

### Ingest Pipeline 和 Logstash 的对比 ###

<table> 
 <thead> 
  <tr> 
   <th>对比</th> 
   <th>Logstash</th> 
   <th>Ingest Pipeline</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>数据输入与输出</td> 
   <td>支持从不同的数据源读取, 并写入不同的数据源</td> 
   <td>支持从 ES REST API 获取数据, 并且写入 Elasticsearch</td> 
  </tr> 
  <tr> 
   <td>数据缓冲</td> 
   <td>实现了简单的数据队列, 支持重写</td> 
   <td>不支持缓冲</td> 
  </tr> 
  <tr> 
   <td>数据处理</td> 
   <td>支持大量的插件, 也支持定制开发</td> 
   <td>内置的插件, 可以开发 Plugin 进行扩展（Plugin 更新需要重启）</td> 
  </tr> 
  <tr> 
   <td>配置和使用</td> 
   <td>增加了一定的架构复杂度</td> 
   <td>无需额外部署</td> 
  </tr> 
 </tbody> 
</table>

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200115211514877.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh,size_16,color_FFFFFF,t_70