怎么使用enrich processor
这篇文章主要介绍"怎么使用enrich processor"，在日常操作中，相信很多人在怎么使用enrich processor问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答"怎么使用enrich processor"的疑惑有所帮助！接下来，请跟着小编一起来学习吧！
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 的内容如下：
| loc | num | company |
|---|---|---|
| 广东省 | A1001 | 腾讯 |
| 上海市 | B1001 | Bilibili |
| 浙江省 | C1001 | 阿里巴巴 |
incoming document 传入的文档如下，通过 num 字段查到对应 source index 中的 loc 的值，添加到 incoming document 中新增 enrich_loc 属性中。
| num | company |
|---|---|
| A1001 | 腾讯 |
| B1001 | Bilibili |
| C1001 | 阿里巴巴 |
第一步：添加 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 pipelinePUT 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 templatePUT _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"的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注网站，小编会继续努力为大家带来更多实用的文章！
