写在前面
如果是我们需要在写入文档或者是返回文档时,进行修改字段值,或者增加字段等操作时,就可以考虑使用ingest pipeline和painless script。如下的需求:
1:ingest pipeline
在es 5中引入了一种新的节点类型ingest node,ingest node具备对数据预处理的能力,使用的就是ingest pipeline。ingest pipeline可以拦截index或者时bulk的请求,对数据处理后重新返回回去,最终保存通过ingest pipeline处理后的数据,如下图:
通过一组processor来处理数据,可以使用内置的processor,也可以使用插件的方式来自定义processor,其中内置的processor列表如下:
另外es提供了simulate的API来模拟ingest pipeline的执行效果:
接下来我们看实际的例子。
1.1:实例
- 定义一个pipeline,方便使用
当然也可以直接写pipeline来用。
# 为ES添加一个 Pipeline
PUT _ingest/pipeline/blog_pipeline
{
"description": "a blog pipeline",
"processors": [
{
"split": {
"field": "tags",
"separator": ","
}
},
{
"set": {
"field": "views",
"value": 0
}
}
]
}
通过slit分割tags,通过set给文档增加一个字段views,值为0。
- 查看定义的pipeline
- 测试pipeline
正式使用之前先来测试一下:
- 不使用pipeline新增数据
PUT tech_blogs/_doc/1
{
"title": "Introducing big data......",
"tags": "hadoop,elasticsearch,spark",
"content": "You konw, for big data"
}
- 使用pipeline新增数据
PUT tech_blogs/_doc/2?pipeline=blog_pipeline
{
"title": "Introducing cloud computering",
"tags": "openstack,k8s",
"content": "You konw, for cloud"
}
看下新增的数据:
- update_by_query使用pipeline重建本索引
因为pipeline处理后插入的数据底层是ArrayList类型,无法再执行split操作,想要重建索引的话,只需要将已经被pipeline处理过数据过滤掉即可,如果是包含views就是处理过的,如下:
POST tech_blogs/_update_by_query?pipeline=blog_pipeline
{
"query": {
"bool": {
"must_not": {
"exists": {
"field": "views"
}
}
}
}
}
1.2:和logstash的比较
2:painless script
palinless script载es5引入,支持Java,其可实现功能如下:
1:对写入的文档增加,删除,修改字段,可配合ingest pipeline使用
2:对返回的文档增加,删除,修改字段(此时不会影响数据库中的文档)
3:function score,基于相关度评分,按照新规则生成新评分,从而影响最终文档的排序结果:https://dongyunqi.blog.csdn.net/article/details/136341656
2.1:painless 上下文
需要需要在painless脚本中访问文档数据,执行具体的操作,所以就需要对文档的访问,而对文档的访问在不同的上下文中是不一样的,主要有如下三种:
1:在pipeline中,通过ctx.fieldname的方式来访问
2:在update中通过ctx_source.fieldname的方式来访问
3:在search和aggregation中通过dock[fieldname]的方式来访问
2.2:实例
2.2.1:pipeline
使用split
processcor将字符串分割为字符串,使用painless scrit增加属性content_length,使用set
processor增加属性views并设置值为0。
POST _ingest/pipeline/_simulate
{
"pipeline": {
"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;
}
"""
}
},
{
"set":{
"field": "views",
"value": 0
}
}
]
},
"docs": [
{
"_index":"index",
"_id":"id",
"_source":{
"title":"Introducing big data......",
"tags":"hadoop,elasticsearch,spark",
"content":"You konw, for big data"
}
},
{
"_index":"index",
"_id":"idxx",
"_source":{
"title":"Introducing cloud computering",
"tags":"openstack,k8s",
"content":"You konw, for cloud"
}
}
]
}
2.2.2:update
- 准备测试数据
DELETE tech_blogs
PUT tech_blogs/_doc/1
{
"title":"Introducing big data......",
"tags":"hadoop,elasticsearch,spark",
"content":"You konw, for big data",
"views":0
}
- 将views增加100
POST tech_blogs/_update/1
{
"script": {
"source": "ctx._source.views += params.new_views",
"params": {
"new_views": 100
}
}
}
GET tech_blogs/_search
- 使用预定义painless脚本的方式更新
预定义painless脚本,并通过入参的方式,给views+1000。
#保存脚本在 Cluster State
POST _scripts/update_views
{
"script": {
"lang": "painless",
"source": "ctx._source.views += params.new_views"
}
}
POST tech_blogs/_update/1
{
"script": {
"id": "update_views",
"params": {
"new_views": 1000
}
}
}
GET tech_blogs/_search
2.2.3:search
返回结果中增加rnd_views
字段,其中以views字段并给其加上1000以内的随机数获取作为值:
GET tech_blogs/_search
{
"script_fields": {
"rnd_views": {
"script": {
"lang": "painless",
"source": """
java.util.Random rnd = new Random();
doc['views'].value+rnd.nextInt(1000);
"""
}
}
},
"query": {
"match_all": {}
}
}