es插⼊数据_ES+Springboot的正确姿势
前⾔
在前边我们探讨了ES的基本概念以及根据不同的场景选择数据迁移的⽅案。在这⼀篇我们来探讨如何与Spring boot集成，以及为了平滑地从Mysql迁移到ES中我们如何”翻译SQL“。
⼀、Spring Boot集成ES
第⼀步我们就要实现Spring boot和ES集成，在Spring boot中主要有Java REST Client、spring-data-elasticarch两种⽅式，这⾥我建议使⽤Elasticarch官⽅提供的Java High Level REST Client来集成，也⽅便在⽣产环境中使⽤阿⾥云的ES云服务。关键的版本信息如下：
ES集群：7.3.0
ES相关依赖：7.3.0
这⾥有两点需要注意：
High Level Client能够向上兼容，例如7.3.0版本的Java High Level REST Client能确保与⼤于等于7.3.0
版本的Elasticarch集群通信。为了保证最⼤程度地使⽤最新版客户端的特性，推荐High Level Client版本与集群版本⼀致。
在集成的过程中可能会踩到⼀些坑，因为Spring Boot的版本、ES集群的版本、High Level Client的版本之间会存在”关联关系“，所以当Demo⽆法正常跑起来的时候能做的就是多尝试⼀些High Level Client版本。
1、pom依赖
org.elasticarch.clientelasticarch-rest-high-level-client7.3.0
org.elasticarchelasticarch7.3.0
org.elasticarch.clientelasticarch-rest-client7.3.0
org.elasticarch.pluginrank-eval-client7.3.0
org.elasticarch.pluginlang-mustache-client7.3.0
女名字
复制代码
2、初始化客户端
@Configuration
public class EsConfig {
@Value("${elasticarch.host}")
public String host;
/**
* 之前使⽤transport的接⼝的时候是9300端⼝，现在使⽤HighLevelClient则是9200端⼝
*/
@Value("${elasticarch.port:9200}")
public int port;
精雕细刻的反义词public static final String SCHEME = "http";
@Value("${elasticarch.urname:admin}")
public String urname;
@Value("${elasticarch.authenticationPassword}")
public String authenticationPassword;豆芽菜图片
@Bean(name = "remoteHighLevelClient")
public RestHighLevelClient restHighLevelClient() {
final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
credentialsProvider.tCredentials(AuthScope.ANY, new UrnamePasswordCredentials(urname,
authenticationPassword));
RestClientBuilder builder = RestClient.builder(new HttpHost(host, port, SCHEME)).
tHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder
.
tDefaultCredentialsProvider(credentialsProvider));
return new RestHighLevelClient(builder);
}
}
复制代码
在上边的代码中需要注意urname和authenticationPassword的认证信息都是在Kibana中设置的。
⼆、Java API
下⾯的代码⽚段均能在单元测试中正常运⾏，在执⾏下边的单元测试之前，我们先创建⼀个_template，⼤家可以选择在Kibana提供的Dev Tools⾥边执⾏。
PUT _template/hero_template
{
"index_patterns":[
"hero*"
],
"mappings":{
"properties":{
"@timestamp":{
"type":"date"
},
"id":{
"type":"integer"
},
"name":{
"type":"keyword"
},
"country":{
"type":"keyword"
},
"birthday":{
"type":"keyword"
},
"longevity":{
"type":"integer"
}
}
}
}
复制代码
1、创建索引
@Test
public void createIndex() throws IOException {
IndexRequest request = new IndexRequest("hero");
request.id("1");
Map map = new HashMap<>();
map.put("id", "1");鸽子汤的家常做法
map.put("name", "曹操");
map.put("country", "魏");
map.put("birthday", "公元155年");
map.put("longevity", "65");
request.source(map);
IndexRespon indexRespon = client.index(request, RequestOptions.DEFAULT);
long version = Version();
asrtEquals(DocWriteRespon.Result.CREATED, Result());
asrtEquals(1, version);
}
复制代码
在ES中索引是我们存储、查询数据的逻辑单元，在ES7.0之后对应的是Mysql中表的概念。上边的代码我们创建了⼀个名为hero的索引，然后我们创建⼀个map作为我们插⼊的第⼀条数据，然后设置到IndexRequest请求对象中。
2、批量插⼊
public void bulkRequestTest() throws IOException {
BulkRequest request = new BulkRequest();
request.add(new IndexRequest("hero").id("2")
.source(XContentType.JSON,"id", "2", "name", "刘备", "country", "蜀", "birthday", "公元161年", "longevity", "61")); request.add(new IndexRequest("hero").id("3")
.source(XContentType.JSON,"id", "3", "name", "孙权", "country", "吴", "birthday", "公元182年", "longevity", "61")); request.add(new IndexRequest("hero").id("4")
.source(XContentType.JSON,"id", "4", "name", "诸葛亮", "country", "蜀", "birthday", "公元181年", "longevity", "53")); request.add(new IndexRequest("hero").id("5")
.source(XContentType.JSON,"id", "5", "name", "司马懿", "country", "魏", "birthday", "公元179年", "longevity", "72")); request.add(new IndexRequest("hero").id("6")
惨不忍睹的意思.source(XContentType.JSON,"id", "6", "name", "荀彧", "country", "魏", "birthday", "公元163年", "longevity", "49")); request.add(new IndexRequest("hero").id("7")
.source(XContentType.JSON,"id", "7", "name", "关⽻", "country", "蜀", "birthday", "公元160年", "longevity", "60")); request.add(new IndexRequest("hero").id("8")
.source(XContentType.JSON,"id", "8", "name", "周瑜", "country", "吴", "birthday", "公元175年", "longevity", "35")); BulkRespon bulkRespon = client.bulk(request, RequestOptions.DEFAULT);
asrtFal(bulkRespon.hasFailures());
}
复制代码
在kibana中查询到的数据如下图
我们后边的查询、更新等等操作都是基于这⾥的数据。
3、更新数据
@Test
public void updateTest() throws IOException {
Map jsonMap = new HashMap<>();
jsonMap.put("country", "魏");
UpdateRequest request = new UpdateRequest("hero", "7").doc(jsonMap);
UpdateRespon updateRespon = client.update(request, RequestOptions.DEFAULT);
asrtEquals(DocWriteRespon.Result.UPDATED, Result());
}
复制代码云南丽江旅游景点
上边的代码如果⽤SQL来表⽰就是下边这样
> update hero t country='魏' where id=7;
复制代码
4、插⼊/更新数据
public void inrtOrUpdateOne(){
Hero hero = new Hero();
hero.tId(5);
hero.tName("曹丕");
hero.tCountry("魏");
hero.tBirthday("公元187年");
hero.tLongevity(39);
IndexRequest request = new IndexRequest("hero");
request.Id().toString());
request.JSONString(hero), XContentType.JSON);
try {
默默奉献IndexRespon indexRespon = client.index(request, RequestOptions.DEFAULT); // 1
asrtEquals(DocWriteRespon.Result.UPDATED, Result());
冰草怎么种植
} catch (Exception e) {
throw new RuntimeException(e);
}
}
复制代码
注意在上边代码中标注1的这⾏代码，是不是和前边创建索引很像？这⾥使⽤⽅法index()我们可以轻松的实现创建索引、插⼊数据、更新数据于⼀体，当指定的索引不存在时即创建索引，当数据不存在时就插⼊，数据存在时就更新。
5、删除数据
@Test
public void deleteByIdTest() throws IOException {
DeleteRequest deleteRequest = new DeleteRequest("hero");
deleteRequest.id("1");
DeleteRespon deleteRespon = client.delete(deleteRequest, RequestOptions.DEFAULT);
asrtEquals(DocWriteRespon.Result.DELETED, Result());
}
复制代码
上边我们删除了在前边创建id=1的数据，其对应的SQL如下：
> delete from hero where id=1;
复制代码
当然，在ES中我们不仅仅可以使⽤主键来删除，我们还可以通过其他的字段条件来删除。
@Test
public void deleteByQueryRequestTest() throws IOException {
DeleteByQueryRequest request = new DeleteByQueryRequest("hero");
request.tConflicts("proceed");
request.tQuery(new TermQueryBuilder("country", "吴"));
BulkByScrollRespon bulkRespon =
client.deleteByQuery(request, RequestOptions.DEFAULT);
asrtEquals(0, BulkFailures().size());
}
复制代码
对应的SQL：
> delete from hero where country='吴';
复制代码
