常州建设局职称网站,做网站什么码,课题组网站建设,防封电销系统简介
我们在使用ES的时候#xff0c;经常的用法就是把其他数据源比如Mysql的数据灌到ES中。
借用ES的一些功能来提供数据的全文检索以及聚合分析之类的功能。
在这个灌数据的过程中#xff0c;我们经常会对数据做一些治理#xff0c;类似ETL的能力。然后把治理后的数据写入…简介
我们在使用ES的时候经常的用法就是把其他数据源比如Mysql的数据灌到ES中。
借用ES的一些功能来提供数据的全文检索以及聚合分析之类的功能。
在这个灌数据的过程中我们经常会对数据做一些治理类似ETL的能力。然后把治理后的数据写入ES中。
我们当然可以自己监听MQ的数据在程序中编写代码来实现这个过程。
也可以使用logstash等中间件来实现这个功能。
这些都可以。但是ES中本身也是提供了一个能力来实现这个功能的。
而且基于这个能力你甚至可以扩展出其他的操作。
这个能力就是Ingest pipelines(摄取管道)。这个系列我们就来看一下摄取管道的能力并且我会分三个章节来完成。这个章节我们只来看一些他的基本用法后面我们会依次讲解 enrich processor和自己开发插件实现管道。
一、什么是Ingest pipelines(摄取管道)
我们先来看官方文档的一些信息。本文使用的ES版本为7.17.7。官方文档的位置位于Ingest pipelines官方文档
# 英文
Ingest pipelines let you perform common transformations on your data before indexing.
For example, you can use pipelines to remove fields, extract values from text, and enrich your data.A pipeline consists of a series of configurable tasks called processors.
Each processor runs sequentially, making specific changes to incoming documents.
After the processors have run, Elasticsearch adds the transformed documents to your data stream or index.# 中文
采集管道可让您在索引之前对数据执行常见的转换。例如您可以使用管道删除字段、从文本中提取值以及丰富数据。
管道由一系列可配置的任务称为处理器组成。每个处理器按顺序运行对传入的文档进行特定更改。
处理器运行后Elasticsearch 会将转换后的文档添加到您的数据流或索引中。而且官方还有一个图示 我们从描述和图示可以看到他很像我们在设计模式中说的责任链设计模式把一个数据从源端拿到之后在写入目标端的索引之前经过一系列的processor处理器最终得到目标数据写入ES索引中。 实际上他也就是这个功能。
而且你可以使用 Kibana 的摄取管道功能或摄取 API 创建和管理摄取管道。 Elasticsearch 将管道存储在集群状态中。我们说基本上ES中的一些操作你都可以通过kibana界面或者是DSL命令来实现。他们是等价的只是一个有界面点点点一个是写代码。作为一个开发我们当然要用命令这种高大上的东西来实现了界面点点点还是让那些小白去点吧没有任何逼格。 而且命令的官方地址位于Ingest pipelines操作命令。 并且我们说这个管道你建立之后如果你想查看你可以去集群状态中去查看具体的操作步骤位于ES查看集群状态
OK至此我们已经知道了他的一些概念以及如何去在文档中定位他。下面我们就来看看如何使用。
二、Ingest pipelines基本步骤
1、节点角色
这个后面我会单独写一篇文章来说这个事这里先提一下如果你想使用Ingest pipelines功能。要求你的节点角色必须是Ingest我们这次不纠结这个因为所有的节点启动默认就有这个角色。我们这次的重点不在节点角色上。 所以我们这个问题目前不存在。
2、操作步骤
我们先来看看他的操作流程然后我们再探究那些深入的。
2.1、创建一个索引写入数据
我们先创建一个索引用于操作。
# 我们创建一个索引他有两个字段一个是text类型的message一个是类似数组类型的tags
# 这里之所以说类似数组是因为ES没有专门的数组类型这个我们后面说。
PUT my-index-01
{settings: {number_of_replicas: 0 # 我是单节点设置副本为0,}, mappings: {properties: {message:{type: text},tags:{type: keyword}}}
}# 然后往里写两条数据
PUT my-index-01/_doc/1
{message: 手机,tags: [ xiao mi , hua wei ]
}PUT my-index-01/_doc/2
{message: 语言,tags: [ java , python ]
}# 查到此时的数据为我们看到数据没毛病
GET my-index-01/_search
{_index : my-index-01,_type : _doc,_id : 1,_score : 1.0,_source : {message : 手机,tags : [xiao mi , hua wei]}
},
{_index : my-index-01,_type : _doc,_id : 2,_score : 1.0,_source : {message : 语言,tags : [java , python]}
}但是我们发现我写进去的数据的tags字段有的是后面有空格有的是前面有空格反正就是对不上。所以我们要把他处理掉。
2.2、创建摄取管道 PUT _ingest/pipeline/my-pipeline-01
{description: 我的第一个pipeline,processors: [{foreach: {field: tags,processor: {trim: {field: _ingest._value}}}}]
}我们看到我创建了一个摄取管道其中processors部分可以传入多个processor如同我们前面说的责任链。这里每一个processor都是责任链中的一个处理节点。 我们加入了第一个processor叫做foreach用来遍历我们的每一条数据的每一个tags字段然后他又嵌套了一个processor叫做trim就是对于每个tags的遍历的属性都做去除前后空格的操作trim在java也常见。而_ingest._value是固定写法表示的就是每一个tags中的元素。取出来做trim。 这就是一个摄取管道的最基本的要素。 这个DSL如果你觉得抽象我可以用java伪代码来表示一下。其实就干了这么点事。
public static void main(String[] args) {ListMyIndex01 list Arrays.asList(new MyIndex01(手机, Arrays.asList(xiao mi , hua wei)), new MyIndex01(语言, Arrays.asList(java , python)));for (MyIndex01 myIndex01 : list) {ListString tags new ArrayList();for (String tag : myIndex01.getTags()) {tags.add(tag.trim());}myIndex01.setTags(tags);}
}Data
AllArgsConstructor
NoArgsConstructor
static class MyIndex01 {private String message;private ListString tags;}2.2、创建一个目标索引
我们创建目标索引用来清洗我们的数据之后存进去。其实和源索引一样。
PUT my-index-01_target
{settings: {number_of_replicas: 0}, mappings: {properties: {message:{type: text},tags:{type: keyword}}}
}2.3、pipelinereindex清洗数据
下面我们就执行reindex把我们源端索引的带着空格的不合理的数据通过摄取管道转换到目标索引中。
POST _reindex
{source: { index: my-index-01},dest: {index: my-index-01_target,pipeline: my-pipeline-01 # 在这里指定我们的摄取管道。}
}执行之后我们可以来查看一下我们清洗之后写入目标索引my-index-01_target的数据。
GET my-index-01_target/_search
此时我们就看到,tags里面数据的前后的空格都被去掉了。
{_index : my-index-01_target,_type : _doc,_id : 1,_score : 1.0,_source : {message : 手机,tags : [xiao mi,hua wei]}
},
{_index : my-index-01_target,_type : _doc,_id : 2,_score : 1.0,_source : {message : 语言,tags : [java,python]}
}OK基于以上你大概知道了摄取管道如何使用但是这只是他的一种使用方式就是pipelinereindex可以从给reindex打配合让数据从一个索引经过摄取管道清洗然后转移到另一个索引。
2.4、直接写入
还记得我们一开始说的那个话题吗我们在写入数据的时候就想清洗之后再写入而不是写完了再转移。这时候pipeline也可以使用。
POST my-index-01/_doc?pipelinemy-pipeline-01
{message: 电脑,tags: [ weiruan , pingguo ]
}我们在写入数据的时候可以直接作用不需要reindex。完全取决于你的场景。 我们来查看一下这个数据我们看到没毛病。
{_index : my-index-01,_type : _doc,_id : q0qiEJEBqSzQT13RTnJx,_score : 1.0,_source : {message : 电脑,tags : [weiruan,pingguo]}
}其余诸如_update_by_queryupdate数据都可以配合使用。具体都可以参考官网。
2.5、查看摄取管道
见操作文档摄取管道操作
2.6、修改摄取管道
修改很简单你原来是啥样的然后改一下在创建一遍他就会把你同名的摄取管道给覆盖掉达到修改的目的。
2.7、删除摄取管道
见操作文档摄取管道操作
2.8、测试摄取管道
这个我单独说一下他的作用就是一个测试而不真正给你执行创建就是模拟了一下但是没有实际执行。 位置位于Simulate pipeline API
三、内置管道
es内部为我们内置了很多管道我们可以开箱即用实际上上文中我们使用的foreach和trim本身就是ES内置的摄取管道。 而这些内置管道可以参考文档位置为ES内置摄取管道文档位置 从图中可以看到有这么多这我还没截全了。实际上比图中的要多。那么我们这里来尝试几个比较常用的其余的你看文档可以自己上手。
0、Foreach processor
这个我们上面说过了就是一个遍历数组元素的作用。
1、Append processor
Append processor文档位置 我们来看官网对他的介绍
Appends one or more values to an existing array if the field already exists and it is an array.
Converts a scalar to an array and appends one or more values to it if the field exists and it is a scalar.
Creates an array containing the provided valuesif the field doesn’t exist. Accepts a single value or an array of values.文档的大意就是如果字段已存在并且是数组则将一个或多个值追加到现有数组。
将标量转换为数组如果该字段存在并且它是标量则向其中追加一个或多个值。
如果该字段不存在则创建一个包含提供的值的数组。接受单个值或值数组。怎么解释呢官网永远不说人话我们来解释一下。 数组就是我们前面用的那个tags那种类型。 标量就是非复杂类型的比如字符串这种。 所以翻译一下这段话的意思就是 1、如果你索引里面存在这个字段了那么就看你这个字段是啥类型的。 如果是数组类型那么Append processor的作用就是给你这个数组add一个值进去。 如果不是数组类型是简单类型那就会把你的这个简单类型转换成数组类型。然后再给你添加进去。
2、如果你索引里面不存在我操作这个字段那直接就是给你创建成数组类型的写进去的也是数组格式。
于是你有了这个概念之后我们就来试试是不是这样呢。
我们还用我们之前的索引my-index-01
PUT my-index-01
{settings: {number_of_replicas: 0}, mappings: {properties: {message:{type: text},tags:{type: keyword}}}
}我们来看一下这个管道的语法如下Append processor
{append: {field: tags,value: [production, {{{app}}}, {{{owner}}}]}
}我来解释一下“field”: “tags”,就是你要操作的字段。你要给哪个字段添加内容这里就写哪个字段。 “value”: [“production”, “{{{app}}}”, “{{{owner}}}”]这一句里面有两种属性一个是production,表示你要给每个tags里面添加一个production字符串而{{{app}}}“这种语法表示你要把你索引里面的app字段的内容也添加进去。如果没有这个app字段那添加进去的就是个空字符串”。 我们来试试。
1.1、给数组字段使用Append processor
PUT _ingest/pipeline/my-pipeline-01
{description: 我的第一个pipeline,processors: [{append: {field: tags,value: [levi,{{{message}}}]}}]
}我们给tags这个数组类型的字段添加内容添加的值是字符串levi同时我们还给tags字段添加了message属性的值。 此时我们写入一条数据。我们预期的就是我们这条数据加进去之后tags里面除了本身的wuya, “zhuomuniao” 还有我们摄取管道添加的levi以及他自己的message字段的值鸟类。
POST my-index-01/_doc?pipelinemy-pipeline-01
{message: 鸟类,tags: [ wuya, zhuomuniao ]
}我们来看一下这个数据。发现没毛病。
1.2、给标量字段使用Append processor
标量字段就是我们的message这个字符串字段。我们按照上面的理论可以知道当我们写入的时候他会把我们这个字符串字段转变为数组类型。并且把值写进数组。
PUT _ingest/pipeline/my-pipeline-02
{description: 我的第二个pipeline,processors: [{append: {field: message, # 操作的标量字段value: [levi,{{{message}}}]}}]
}我们这里是给message字段添加值而且添加的内容是levi字符串以及message本身这个内容等于是把message添加两次。 POST my-index-01/_doc?pipelinemy-pipeline-02
{message: 人类,tags: [ man, woman ]
} 我们看到他把我们的message转成数组了。而且确实message里面的人类被写进去两次。
1.3、Append processor操作一个不存在的字段
PUT _ingest/pipeline/my-pipeline-03
{description: 我的第三个pipeline,processors: [{append: {field: message,value: [production,{{{123}}}]}}]
}我们操作了一个不存在的字段123这时候我们看看会发生什么。 其实还是符合我们的预期的。
2、Date processor
Date processor文档位置 我们来看官网对他的介绍
Parses dates from fields, and then uses the date or timestamp as the timestamp for the document.
By default, the date processor adds the parsed date as a new field called timestamp.
You can specify a different field by setting the target_field configuration parameter.
Multiple date formats are supported as part of the same date processor definition.
They will be used sequentially to attempt parsing the date field, in the same order they were defined as part of the processor definition.意思就是
解析字段中的日期然后使用日期或时间戳作为文档的时间戳。
默认情况下日期处理器将解析的日期添加为名为 timestamp 的新字段。
您可以通过设置 target_field 配置参数来指定不同的字段。支持多种日期格式作为同一日期处理器定义的一部分。
它们将按顺序使用来尝试解析日期字段其顺序与它们被定义为处理器定义的一部分的顺序相同。换句话说就是他会把你数据中的某个表示日期的字段给解析成为一个你指定的日期格式的值
然后赋值给你指定的那个字段。如果你没指定这个字段那就他会给你自己创建一个叫做timestamp的字段。
并且这个转换的日期格式可以是多种实际上我们一般只需要一种整个系统应该统一日期格式不然未来
整出什么花活来就难受了。3、Set processor
Set processor文档位置 我们来看官网对他的介绍
Sets one field and associates it with the specified value. If the field already exists,its value will be replaced with the provided one.意思就是
设置一个字段并将其与指定值关联。如果该字段已存在则其值将替换为提供的值。
换句话说就是为某一个字段设置值这个值可以是你指定写死的也可以是来自索引中的某个其他字段。3.1、创建一个索引
PUT my-index-01
{settings: {number_of_replicas: 0}, mappings: {properties: {firstname:{type: keyword},lastname:{type: keyword},fullname:{type: keyword}}}
}3.2、创建set processor
PUT _ingest/pipeline/my-set-pipeline-01
{description: 我的第一个set pipeline,processors: [{set: {field: fullname,value: {{{lastname}}} {{{firstname}}} ok}}]
}我们为我们的fullname字段通过set processor设置了值值的内容分别是文档中的lastname字段和firstname字段的值拼起来的。并且我还在后面固定了一个字符串ok。
3.3、通过摄取管道写入数据
POST my-index-01/_doc?pipelinemy-set-pipeline-01
{firstname: levi,lastname: jack
}查看结果符合我们的预期。
4、Script processor
这是我们最后要介绍的一个摄取管道其余的后面有时间慢慢补上这个比较强大我们就来看看这个。 Script processor官方文档 我们来看官网对他的介绍
Runs an inline or stored script on incoming documents. The script runs in the ingest context.The script processor uses the script cache to avoid recompiling the script for each incoming document.
To improve performance, ensure the script cache is properly sized before using a script processor in production.意思就是
对传入文档运行内联或存储的脚本。该脚本在摄取管道上下文中运行。
脚本处理器使用脚本缓存来避免为每个传入文档重新编译脚本。
为了提高性能请在生产中使用脚本处理器之前确保脚本缓存的大小正确。换句话说以前我们那些管道都是一些内置的你无外乎就是配置一些参数但是这个不一样这个你可以编辑
一些脚本逻辑来控制这个生效逻辑。他的可选参数如下其实其他管道也有可选参数大差不差之所以前面的没写是因为不难看文档都能看懂这个我这里写一下。 我们先来使用一下实际上这里很多东西都和脚本有关具体可以参考文档。
4.1、创建一个索引
PUT my-index-01
{settings: {number_of_replicas: 0}, mappings: {properties: {tags:{type: keyword}}}
}索引很简单就是一个tags我们里面准备存一些数组类型。
4.2、创建摄取管道
PUT _ingest/pipeline/my-script-pipeline-01
{description: 我的第一个script pipeline,processors: [{script: {description: 把tags中包含huawei的元素移除,lang: painless,source: def tags ctx[tags];if(tags ! null){for (int i 0; i tags.length; i) {if(tags[i] params.removetag){tags.remove(i);}}},params: {removetag: huawei}}}]
}你可以看到我们这个脚本实际上和java很像他的ctx就是你每条数据过管道的时候他获取到这个数据上下文然后取出你的tags字段遍历发现和参数一样的就移除。
4.3、通过摄取管道写入数据
POST my-index-01/_doc?pipelinemy-script-pipeline-01
{tags: [xiaomi,huawei,meizu]
}查看数据我们看到huawei确实被移除了。
四、来个需求
1、需求梳理
在这个实操中我们会用java结合DSL的方式来实现实际开发中如何使用摄取管道的操作。
假如此时我们有个需求客户提供了一个数据表是mysql表叫做newstable领导要你把这个表接进来然后写入你的ES。 这个表就两个字段一个是name一个是tags
但是呢他的表里面 name字段有的前后有空格有的还各种大小写不统一领导要你都变成小写。 最离谱的是name字段里面的内容有html标签这你完全不能忍展示出来成啥了。 tags字段你想在你的ES里面存成数组结果mysql还是用逗号隔开的字符串。 而且tags字段数据是有一些敏感词对于你们公司来说tnd这属于敏感词你要给他过滤掉。
基于这样的脏数据你要在数据进入ES之前给他治理一波。 开发之前我们先来梳理一下这些需求的点
1、前后有空格问题不大我们用trim processor来去掉。
2、大小写不统一都要变为小写。问题也不大我们用Lowercase processor来处理。
3、还有一个tags字段你要存数组但是源端是逗号隔开的字符串你要切割开我们使用Split processor。
4、html标签这个问题也不大我们用HTML strip processor来处理。
5、敏感词过滤我们用Script processor来处理移除掉。于是我们选好了我们使用的管道之后我们就来进行开发我的环境很简单。
2、工程环境
1、jdk1.8
2、springboot2.3.4.RELEASE
3、maven 3.6.1
4、elasticsearch 7.17.7以下是我的pom依赖配置。
?xml version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersionparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.3.4.RELEASE/version/parentgroupIdcom.leiv/groupIdartifactIdspringboot-demo/artifactIdversion0.0.1-SNAPSHOT/versionnamespringboot-demo/namedescriptionspringboot-demo/descriptionpackagingjar/packagingpropertiesjava.version1.8/java.versionelasticsearch.version7.17.7/elasticsearch.version/propertiesdependencies!-- web场景启动器 --dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdoptionaltrue/optional/dependency!--引入ES客户端依赖--dependencygroupIdorg.elasticsearch.client/groupIdartifactIdelasticsearch-rest-client/artifactIdversion${elasticsearch.version}/version/dependencydependencygroupIdorg.elasticsearch/groupIdartifactIdelasticsearch/artifactIdversion${elasticsearch.version}/version/dependencydependencygroupIdorg.elasticsearch.client/groupIdartifactIdelasticsearch-rest-high-level-client/artifactIdversion${elasticsearch.version}/version/dependency/dependenciesbuildpluginsplugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactId/plugin/plugins/build/project
我的yaml文件配置。
elasticsearch:nodes: 127.0.0.1:9200cluster-name: my-applicationcluster-port: 9300好了我们已经有了环境就可以进行开发了。
3、代码编写
3.1、es客户端类配置
/*** 方法描述:es基础配置绑定,映射配置文件*/
Data
ConfigurationProperties(prefix elasticsearch)
public class EsConfigProperties {private String clusterName;private String[] nodes;private Integer clusterPort;
}Slf4j
Configuration
EnableConfigurationProperties(EsConfigProperties.class)
public class ElasticsearchConfiguration {private EsConfigProperties esConfigProperties;public ElasticsearchConfiguration(EsConfigProperties esConfigProperties) {this.esConfigProperties esConfigProperties;}/*** 方法描述初始化es客户端 因为我们还没做安全管理所以这里使用无密码的客户端* return*/Bean(destroyMethod close,name restHighLevelClient)public RestHighLevelClient initRestClient() {RestHighLevelClient restHighLevelClient null;try {String[] nodes esConfigProperties.getNodes();if(Objects.isNull(nodes) || nodes.length 0){log.info(es客户端初始化失败未发现es节点配置{}, JSON.toJSONString(nodes));}ListHttpHost hostList new ArrayList();for (String node : nodes) {String[] nodeAttr node.split(:);HttpHost httpHost new HttpHost(nodeAttr[0], Integer.parseInt(nodeAttr[1]));hostList.add(httpHost);}HttpHost [] httpHostArray hostList.toArray(new HttpHost[]{});RestClientBuilder builder RestClient.builder(httpHostArray).setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {Overridepublic HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {return httpAsyncClientBuilder;}});restHighLevelClient new RestHighLevelClient(builder);log.info(ES高级客户端初始化完成......);}catch (Exception e){e.printStackTrace();log.info(ES高级客户端初始化失败{},e.getMessage());}return restHighLevelClient;}/*** 方法描述解决es和netty版本冲突问题*/PostConstructvoid init() {System.setProperty(es.set.netty.runtime.available.processors, false);}}3.2、在kibana创建管道
实际上这种东西你也可以用java客户端来做但是一般我们用脚本创建好保存起来就行java代码有了问题你想改还得发版所以我们选在在kibana端编写DSL。 我们把之前的梳理搬出来看看我们要创建哪几个管道。
1、前后有空格问题不大我们用trim processor来去掉。
2、大小写不统一都要变为小写。问题也不大我们用Lowercase processor来处理。
3、还有一个tags字段你要存数组但是源端是逗号隔开的字符串你要切割开我们使用Split processor。
4、html标签这个问题也不大我们用HTML strip processor来处理。
5、敏感词过滤我们用Script processor来处理移除掉。根据这几个需求我们去挨个创建管道注意哈我们的管道是一个链你完全可以在一个pipline里面配置多个管道他会让你的每条数据挨个去执行这些管道的。pipline翻译就是流水线其实见名知义了。
我们编写我们的管道流水线。注意我们的管道顺序编织是有顺序的我们选择先去除html标签在转小写因为如果你先转小写可能把html标签给转成别的了。这样会影响你的处理所以我们在编写的时候要考虑执行顺序带来的影响。
PUT _ingest/pipeline/my-etl-newstable-pipeline
{description: 我的清洗newstable表数据的pipeline,processors: [{html_strip: {field: name}},{trim: {field: name}},{lowercase: {field: name}},{split: {field: tags,separator: ,,preserve_trailing: true}},{script: {description: 把tags中包含tmd,tnd的元素移除,lang: painless,source: def tags ctx[tags];if(tags ! null){for (int i 0; i tags.length; i) {if(tags[i] params.tndparam){tags.remove(i);}}},params: {tndparam: tnd}}}]
}所以我们最后的处理流程就是 OK,至此我们的管道就创建完毕了。我们接下来模拟在接口中写入数据使用该管道。
3.3、创建索引
创建我们要写入的索引。
PUT newstable-index
{settings: {number_of_replicas: 0}, mappings: {properties: {name:{type: keyword},tags:{type: keyword}}}
}
3.4、创建实体类和接口
// 模拟mysql表的实体映射
Data
AllArgsConstructor
NoArgsConstructor
public class NewsTableEntity {// mysql中的name映射private String name;// mysql中的tags映射private String tags;
}RestController
RequestMapping(/doc/es)
public class EsController {AutowiredRestHighLevelClient restHighLevelClient;// 写入的ES索引名称private static final String INDEX_NAME newstable-index;// 指定pipeline名称这里要和你在kibana创建的pipeline对上private static final String PIPELINE_NAME my-etl-newstable-pipeline;GetMapping(/addDoc)public boolean addDoc() throws IOException {ListNewsTableEntity tableEntities new ArrayList();// 模拟mysql表数据name字段都是有html标签的而且前后都有空格而且tags字段都是逗号分隔的字符串有的还有敏感词tndtableEntities.add(new NewsTableEntity( H1 Hello world!/H1 ,HUAWEI,小米,苹果,tnd));tableEntities.add(new NewsTableEntity( P 你好/P ,su7,bmw));// 这里构造批量请求你单个请求也一样的实际开发一般都是批量性能好一些BulkRequest bulkRequest new BulkRequest();for (NewsTableEntity tableEntity : tableEntities) {IndexRequest request new IndexRequest(INDEX_NAME).source(JSONObject.toJSONString(tableEntity), XContentType.JSON)// 指定pipeline这里只是示例实际开发中根据业务需求来.setPipeline(PIPELINE_NAME);bulkRequest.add(request);}BulkResponse bulkResponse restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);return bulkResponse.hasFailures();}
}3.5、测试接口 查看ES中的数据发现我们已经清洗成功了。name去除了空格和html标签tags变为了数组并且移除了敏感词。
