共计 6183 个字符,预计需要花费 16 分钟才能阅读完成。
这篇文章主要介绍“怎么使用 enrich processor”,在日常操作中,相信很多人在怎么使用 enrich processor 问题上存在疑惑,丸趣 TV 小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”怎么使用 enrich processor”的疑惑有所帮助!接下来,请跟着丸趣 TV 小编一起来学习吧!
enrich processor 简介
ingest pipeline 可以在传入的文档被索引之前,对文档进行预处理,通过 processor 中定义的一系列规则来修改文档的内容(例如大小写转换等)。
在 Elasticsearch 7.5 版本引入了 enrich processor,可以将现有索引(source index)中的数据添加到传入的文档(incoming document)中。
比如,你可以在如下的场景中用到:
根据已知的 IP 地址识别 Web 服务或供应商。
根据产品 ID 将产品信息添加到零售订单中。
根据电子邮件地址补充联系信息。
根据用户坐标添加邮政编码。
使用 enrich processor
使用 enrich processor 有如下几个步骤:
1. 添加 enrich data:添加 document(enrich data)到一个或者多个的 source index 中,这些 document 中应包含之后要添加到 incoming documents 中的数据。
2. 创建 enrich policy:enrich policy 中应至少包含如下参数:
指定 source index 的。
指定 incoming documents 和 source index 用于匹配的属性。
指定要添加到 incoming documents 中的属性。
3. 执行 enrich policy:执行完后会自动创建相应的 enrich index,enrich index 和普通索引不同,进行了优化。
4. 在 ingest pipeline 使用 enrich processor:enrich processor 使用 enrich index 来查询。
背景说明
source index 的内容如下:
locnumcompany 广东省 A1001 腾讯上海市 B1001Bilibili 浙江省 C1001 阿里巴巴
incoming document 传入的文档如下,通过 num 字段查到对应 source index 中的 loc 的值,添加到 incoming document 中新增 enrich_loc 属性中。
numcompanyA1001 腾讯 B1001BilibiliC1001 阿里巴巴第一步:添加 enrich data
通过 _bulk API 批量添加文档到 location 索引,这些文档和普通的文档一样。
POST _bulk
{index : { _index : location}}
{loc : 广东省 , company : 腾讯 , num : A1001}
{index : { _index : location}}
{loc : 上海市 , company : Bilibili , num : B1001}
{index : { _index : location}}
{loc : 浙江省 , company : 阿里巴巴 , num : C1001}
第二步:创建 enrich policy
enrich policy 一旦创建,就不能更新或者修改。
PUT /_enrich/policy/my-policy
match : {
indices : location , #source index 索引名,就是前面创建的 enrich data 对应的索引
match_field : num , #source index 中的属性名,用于 incoming documents 和 source index 匹配的属性,属性名一样都是 num
enrich_fields : [loc], # 添加到 incoming documents 中的属性
# 可选,过滤 source index 的文档,只有 loc.keyword 是上海市的 enrich data 才能将属性添加到 incoming documents 中
query : {
match : {
loc.keyword : 上海市
}
}
}
}
第三步:执行 enrich policy
当创建了 enrich policy 后,你可以通过 execute enrich policy API 去执行 enrich policy。当执行 enrich policy 后,会自动创建 enrich index。
直接将 incoming document 与 source index 中的文档匹配可能会很慢且占用大量资源。为了加快处理速度,enrich processor 使用了 enrich index。enrich index 包含来自 source index 的 enrich data,enrich index 具有一些特殊属性可帮助简化它们:
它们是系统索引,这意味着它们由 Elasticsearch 在内部进行管理,仅适用于 enrich processor。
它们始终以 .enrich- * 开头。
它们是只读的,这意味着你不能直接更改它们。
它们被强制合并以便快速检索。
当 source index 中新增或者修改了数据,只需要重新执行 enrich policy 就可以更改 enrich index,从而更新 enrich processor。
通过以下命令执行 enrich policy:
PUT /_enrich/policy/my-policy/_execute
查看自动创建的 enrich index:
GET _cat/indices/.enrich*
# 返回结果
green open .enrich-my-policy-1616136526661 Vxal9lLBSlKS5lmzMpFfwQ 1 3 1 0 13.4kb 3.3kb
我感觉 enrich policy 这里有个小 bug,当删除 enrich policy 时,例如删除的 enrich policy 为 my-policy-1,会同时删除 my-policy-1 的 enrich index 和 enrich policy,但是如果原先还有个 my-policy-2(两个 enrich policy 在 - 之前是一样的),会把 my-policy-2 的 enrich index 也误删了(enrich policy 不删)。
第四步:在 ingest pipeline 使用 enrich processor
PUT _ingest/pipeline/loc-pipeline
processors : [
{
enrich : {
policy_name : my-policy , # 引用前面创建的 enrich policy
field : num , # incoming document 中的属性名,用于和 source index 中的属性匹配值
# 在 incoming document 中新增的属性, # 包含在 enrich policy 中定义的 match_field 和 enrich_fields 的值
target_field : enrich_loc
}
}
]
}
验证
使用 simulate 用来调试 ingest pipeline 的效果,由于 source index 中匹配到的 loc.keyword 不是上海市,不会对这个文档进行处理:
POST _ingest/pipeline/loc-pipeline/_simulate
docs : [
{
_source : {
num : A1001 ,
company : 腾讯
}
}
]
# 返回结果
docs : [
{
doc : {
_index : _index ,
_type : _doc ,
_id : _id ,
_source : {
company : 腾讯 ,
num : A1001
},
_ingest : {
timestamp : 2021-03-19T06:56:45.754486259Z
}
}
}
]
}
这个文档的 loc.keyword 是上海市,因此会添加上 enrich data 中指定的属性:
POST _ingest/pipeline/loc-pipeline/_simulate
docs : [
{
_source : {
num : B1001 ,
company : Bilibili
}
}
]
# 返回结果
docs : [
{
doc : {
_index : _index ,
_type : _doc ,
_id : _id ,
_source : {
company : Bilibili ,
enrich_loc : {
loc : 上海市 ,
num : B1001
},
num : B1001
},
_ingest : {
timestamp : 2021-03-19T06:56:29.393585306Z
}
}
}
]
}
在 simulate 调试成功之后,我们在插入文档的时候指定 ingest pipeline:
# 方式一:单条插入
POST origin-location/_doc?pipeline=loc-pipeline
num : A1001 ,
company : 腾讯
POST origin-location/_doc?pipeline=loc-pipeline
num : B1001 ,
company : Bilibili
# 方式二:批量插入
POST _bulk?pipeline=loc-pipeline
{index :{ _index : origin-location}}
{num : A1001 , company : 腾讯}
{index :{ _index : origin-location}}
{num : B1001 , company : Bilibili}
查看插入的结果:
GET origin-location/_search
#返回结果
took : 12,
timed_out : false,
_shards : {
total : 1,
successful : 1,
skipped : 0,
failed : 0
},
hits : {
total : {
value : 2,
relation : eq
},
max_score : 1.0,
hits : [
{
_index : origin-location ,
_type : _doc ,
_id : zXxLSXgBUc4opBV-QiOv ,
_score : 1.0,
_source : {
num : A1001 ,
company : 腾讯
}
},
{
_index : origin-location ,
_type : _doc ,
_id : znxLSXgBUc4opBV-SCPk ,
_score : 1.0,
_source : {
num : B1001 ,
company : Bilibili ,
enrich_loc : {
loc : 上海市 ,
num : B1001
}
}
}
]
}
}
也可以指定索引默认使用的 ingest pipeline,这样就不用每次在插入文档的时候指定 ingest pipeline 了:
# 指定索引默认使用的 ingest pipeline
PUT origin-location2
settings : {
default_pipeline : loc-pipeline
}
# 插入数据
POST _bulk
{index :{ _index : origin-location2}}
{num : A1001 , company : 腾讯}
{index :{ _index : origin-location2}}
{num : B1001 , company : Bilibili}
# 查看结果
GET origin-location2/_search
# 输出结果
took : 8,
timed_out : false,
_shards : {
total : 1,
successful : 1,
skipped : 0,
failed : 0
},
hits : {
total : {
value : 2,
relation : eq
},
max_score : 1.0,
hits : [
{
_index : origin-location2 ,
_type : _doc ,
_id : CXxPSXgBUc4opBV-oyTJ ,
_score : 1.0,
_source : {
num : A1001 ,
company : 腾讯
}
},
{
_index : origin-location2 ,
_type : _doc ,
_id : CnxPSXgBUc4opBV-oyTJ ,
_score : 1.0,
_source : {
num : B1001 ,
company : Bilibili ,
enrich_loc : {
loc : 上海市 ,
num : B1001
}
}
}
]
}
}
另外还可以使用 index template,通过正则表达式的方式匹配多个索引,来指定索引使用的 ingest pipeline:
# 使用 index template
PUT _template/my-template
index_patterns : [origin-*],
settings : {
default_pipeline : loc-pipeline
}
# 插入数据
POST _bulk
{index :{ _index : origin-location3}}
{num : A1001 , company : 腾讯}
{index :{ _index : origin-location3}}
{num : B1001 , company : Bilibili}
# 查看结果
GET origin-location3/_search
# 输出结果
took : 2,
timed_out : false,
_shards : {
total : 1,
successful : 1,
skipped : 0,
failed : 0
},
hits : {
total : {
value : 2,
relation : eq
},
max_score : 1.0,
hits : [
{
_index : origin-location3 ,
_type : _doc ,
_id : XnxVSXgBUc4opBV-1yRp ,
_score : 1.0,
_source : {
num : A1001 ,
company : 腾讯
}
},
{
_index : origin-location3 ,
_type : _doc ,
_id : X3xVSXgBUc4opBV-1yRp ,
_score : 1.0,
_source : {
num : B1001 ,
company : Bilibili ,
enrich_loc : {
loc : 上海市 ,
num : B1001
}
}
}
]
}
}
到此,关于“怎么使用 enrich processor”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注丸趣 TV 网站,丸趣 TV 小编会继续努力为大家带来更多实用的文章!