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

本文基于Elasticsearch7.x

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

Ingest Pipeline介绍

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 的对比

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