使用JAVA操作ElasticSearch

一、Spring Data Elasticsearch

Elasticsearch提供的Java客户端有一些不太方便的地方：

• 很多地方需要拼接Json字符串，在java中拼接字符串有多恐怖你应该懂的
• 需要自己把对象序列化为json存储
• 查询到结果也需要自己反序列化为对象

因此，我们这里就不介绍原生的Elasticsearch客户端API了，而是介绍Spring提供的套件：Spring Data Elasticsearch。

1.1、简介

Spring Data Elasticsearch是Spring Data项目下的一个子模块。
查看 Spring Data的官网：http://projects.spring.io/spring-data/

Spring Data 的使命是给各种数据访问提供统一的编程接口，不管是关系型数据库（如MySQL），还是非关系数据库（如Redis），或者类似Elasticsearch这样的索引数据库。从而简化开发人员的代码，提高开发效率。

包含很多不同数据操作的模块：


Spring Data Elasticsearch的页面：
https://projects.spring.io/spring-data-elasticsearch/

特征：

• 支持Spring的基于@Configuration的java配置方式，或者XML配置方式
• 提供了用于操作ES的便捷工具类ElasticsearchTemplate。包括实现文档到POJO之间的自动智能映射。
• 利用Spring的数据转换服务实现的功能丰富的对象映射
• 基于注解的元数据映射方式，而且可扩展以支持更多不同的数据格式
• 根据持久层接口自动生成对应实现方法，无需人工编写基本操作代码（类似mybatis，根据接口自动得到实现）。当然，也支持人工定制查询

二、新建项目

依赖：

<dependency>
	<groupId>org.springframework.data</groupId>
	<artifactId>spring‐data‐elasticsearch</artifactId>
</dependency>
1
2
3
4

实体类：

package com.esearch.maltose.pojo;

import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.Document;
import org.springframework.data.elasticsearch.annotations.Field;
import org.springframework.data.elasticsearch.annotations.FieldType;

/**
 * @Author：sgw(songguowei@bosssoft.com.cn)
 * @DATE 2019/9/5
 * @description：实体类
 */

@Document(indexName = "item", type = "docs", shards = 1, replicas = 0)
public class Item {
    @Id
    Long id;

    /**
     * 标题
     */
    @Field(type = FieldType.Text, analyzer = "ik_max_word")
    String title;
    /**
     * 分类
     */
    @Field(type = FieldType.Keyword)
    String category;

    /**
     * 品牌
     */
    @Field(type = FieldType.Keyword)
    String brand;

    /**
     * 价格
     */
    @Field(type = FieldType.Double)
    Double price;

    /**
     * 图片地址
     */
    @Field(index = false, type = FieldType.Keyword)
    String images;

    public Item(Long id, String title, String category, String brand, Double price, String images) {
        this.id = id;
        this.title = title;
        this.category = category;
        this.brand = brand;
        this.price = price;
        this.images = images;
    }

    public Long getId() {
        return id;
    }

    public void setId(Long id) {
        this.id = id;
    }

    public String getTitle() {
        return title;
    }

    public void setTitle(String title) {
        this.title = title;
    }

    public String getCategory() {
        return category;
    }

    public void setCategory(String category) {
        this.category = category;
    }

    public String getBrand() {
        return brand;
    }

    public void setBrand(String brand) {
        this.brand = brand;
    }

    public Double getPrice() {
        return price;
    }

    public void setPrice(Double price) {
        this.price = price;
    }

    public String getImages() {
        return images;
    }

    public void setImages(String images) {
        this.images = images;
    }

    public Item() {
    }

    @Override
    public String toString() {
        return id+" "+brand+" "+price;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

application.yml:

spring:
  data:
    elasticsearch:
      cluster-name: elasticsearch
      cluster-nodes: 192.168.11.128:9300
1
2
3
4
5

2.2、Template索引操作

2.2.1、新增索引——使用ElasticsearchTemplate

package com.esearch.maltose;

import com.esearch.maltose.pojo.Item;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.test.context.junit4.SpringRunner;

/**
 * @Author：sgw(songguowei@bosssoft.com.cn)
 * @DATE 2019/9/5
 * @description：测试新建索引
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = MaltoseApplication.class)
public class IndexTest {
    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;
    @Test
    public void testCreate(){
        // 创建索引，会根据Item类的@Document注解信息来创建
        elasticsearchTemplate.createIndex(Item.class);
        // 配置映射，会根据Item类中的id、Field等字段来自动完成映射
        elasticsearchTemplate.putMapping(Item.class);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

没有执行测试方法之前：

执行测试方法后：

新增索引——不使用ElasticsearchTemplate

创建数据访问接口

public interface ArticleSearchDao extends ElasticsearchRepository<Item,String> {
}
1
2

创建业务逻辑service类

@
Service
public class ArticleSearchService {
	@Autowired
	private ArticleSearchDao articleSearchDao;
	/**
	* 增加
	* @param article
	*/
	public void add(Item item){
		articleSearchDao.save(item);
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13

Controller：

@RestController
@CrossOrigin
@RequestMapping("/item")
public class ArticleSearchController {
	@Autowired
	private ArticleSearchService articleSearchService;
	@RequestMapping(method= RequestMethod.POST)
	public Result save(@RequestBody Item item){
		articleSearchService.save(item);
		return new Result(true, StatusCode.OK, "操作成功");
	}
}
1
2
3
4
5
6
7
8
9
10
11
12

2.2.2、删除索引

 @Test
 public void testDelete(){
     // 可以根据类名或索引名删除。
     elasticsearchTemplate.deleteIndex(Item.class);
     //elasticsearchTemplate.deleteIndex("item");
    }
1
2
3
4
5
6

2.2.3、搜索

数据访问接口新增方法

public Page<Item> findByTitleOrContentLike(String title, String
content, Pageable pageable);
1
2

Service新增方法

public Page<Item> findByTitleLike(String keywords, int page, int size){
		PageRequest pageRequest = PageRequest.of(page‐1, size);
		return articleSearchRepository.findByTitleOrContentLike(keywords,keywords,
		pageRequest);
}
1
2
3
4
5

Controller方法

@RequestMapping(value="/search/{keywords}/{page}/{size}",method=
RequestMethod.GET)
public Result findByTitleLike(@PathVariable String keywords,
	@PathVariable int page, @PathVariable int size){
	Page<Article> articlePage =articleSearchService.findByTitleLike(keywords,page,size);
	return new Result(true, StatusCode.OK, "查询成功",new PageResult<Item(articlePage.getTotalElements(),
	articlePage.getContent()));
}
1
2
3
4
5
6
7
8

2.3、Repository文档操作

Spring Data 的强大之处，就在于你不用写任何DAO处理，自动根据方法名或类的信息进行CRUD操作。只要你定义一个接口，然后继承Repository提供的一些子接口，就能具备各种基本的CRUD功能。

我们只需要定义接口，然后继承它就OK了。

package com.esearch.maltose.repository;

import com.esearch.maltose.pojo.Item;
import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;

public interface ItemRepository extends ElasticsearchRepository<Item,Long> {
    
}
1
2
3
4
5
6
7
8

2.3.1、新增文档（在表里加数据）

package com.esearch.maltose;

import com.esearch.maltose.pojo.Item;
import com.esearch.maltose.repository.ItemRepository;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

/**
 * @Author：sgw(songguowei@bosssoft.com.cn)
 * @DATE 2019/9/5
 * @description：新增数据
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = MaltoseApplication.class)
public class IndexTest {

    @Autowired
    private ItemRepository itemRepository;

    @Test
    public void index() {
        Item item = new Item(1L, "小米手机7", " 手机","小米", 3499.00, "https://img-blog/data:image");
        itemRepository.save(item);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

查询结果：

2.3.2、批量新增文档（批量在表里加数据）

    @Test
    public void indexList() {
        List<Item> list = new ArrayList<>();
        list.add(new Item(2L, "readMeK20", " 手机", "小米", 3699.00, "http://image.maltose.com/123.jpg"));
        list.add(new Item(3L, "华为META30", " 手机", "华为", 4499.00, "http://image.maltose.com/456.jpg"));
        // 接收对象集合，实现批量新增
        itemRepository.saveAll(list);
    }
1
2
3
4
5
6
7
8

结果：

2.3.3、修改文档

修改和新增是同一个接口，区分的依据就是id，这一点跟我们在页面发起PUT请求是类似的。即ID已经存在的话，就执行修改，ID不存在的话就执行新增；

2.3.4、基本查询

ElasticsearchRepository提供了一些基本的查询方法：

 @Test
    public void queryTest() {
        Iterable<Item> items = itemRepository.findAll(Sort.by(Sort.Direction.DESC, "price"));
        items.forEach(item-> System.out.println(item));
    }
1
2
3
4
5

结果：

2.3.5、自定义方法

Spring Data 的另一个强大功能，是根据方法名称自动实现功能。

比如：你的方法名叫做：findByTitle，那么它就知道你是根据title查询，然后自动帮你完成，无需写实现类。

当然，方法名称要符合一定的约定：

例如，我们来按照价格区间查询，定义这样的一个方法：
接口：

public interface ItemRepository extends ElasticsearchRepository<Item,Long> {
    /**
     * 根据价格区间查询
     * @param price1  价格范围最小值
     * @param price2  价格范围最大值
     * @return
     */
    List<Item> findByPriceBetween(double price1, double price2);
}
1
2
3
4
5
6
7
8
9

测试：

 @Test
    public void queryByPriceBetween(){
        List<Item> list = this.itemRepository.findByPriceBetween(2000.00, 3500.00);
        for (Item item : list) {
            System.out.println("item = " + item);
        }
    }
1
2
3
4
5
6
7

结果：

虽然基本查询和自定义方法已经很强大了，但是如果是复杂查询（模糊、通配符、词条查询等）就显得力不从心了。此时，我们只能使用原生查询。

3、高级查询

3.1、基本查询

    @Test
    public void testQuery(){
        // 词条查询
        MatchQueryBuilder queryBuilder = QueryBuilders.matchQuery("title", "小米");
        // 执行查询
        Iterable<Item> items = this.itemRepository.search(queryBuilder);
        //打印
        items.forEach(System.out::println);
    }
1
2
3
4
5
6
7
8
9

结果：

Repository的search方法需要QueryBuilder参数，elasticSearch为我们提供了一个对象QueryBuilders：

QueryBuilders提供了大量的静态方法，用于生成各种不同类型的查询对象，例如：词条、模糊、通配符等QueryBuilder对象。

elasticsearch提供很多可用的查询方式，但是不够灵活。如果想执行过滤或者聚合查询等就很难了。

3.2、自定义查询

    @Test
    public void testNativeQuery(){
        // 构建查询条件
        NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder();
        // 添加基本的分词查询
        queryBuilder.withQuery(QueryBuilders.matchQuery("title", "小米"));
        // 执行搜索，获取结果
        Page<Item> items = this.itemRepository.search(queryBuilder.build());
        // 打印总条数
        System.out.println("总条数："+items.getTotalElements());
        // 打印总页数
        System.out.println("总页数："+items.getTotalPages());
        //打印详细信息
        items.forEach(System.out::println);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

结果：

NativeSearchQueryBuilder：Spring提供的一个查询条件构建器，帮助构建json格式的请求体

Page：默认是分页查询，因此返回的是一个分页的结果对象，包含属性：

• totalElements：总条数
• totalPages：总页数
• Iterator：迭代器，本身实现了Iterator接口，因此可直接迭代得到当前页的数据
• 其它属性：
  

3.3、分页查询

利用NativeSearchQueryBuilder可以方便的实现分页：

    @Test
    public void testNativeQuery2(){
        // 构建查询条件
        NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder();
        // 添加基本的分词查询
        queryBuilder.withQuery(QueryBuilders.termQuery("category", "手机"));

        // 初始化分页参数
        int page = 0;
        int size = 3;
        // 设置分页参数
        queryBuilder.withPageable(PageRequest.of(page, size));

        // 执行搜索，获取结果
        Page<Item> items = this.itemRepository.search(queryBuilder.build());
        // 打印总条数
        System.out.println("总条数："+items.getTotalElements());
        // 打印总页数
        System.out.println("总页数："+items.getTotalPages());
        // 每页大小
        System.out.println("每页大小："+items.getSize());
        // 当前页
        System.out.println("当前页："+items.getNumber());
        items.forEach(System.out::println);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

结果：

可以发现，Elasticsearch中的分页是从第0页开始。

3.3.4、排序

排序也通过NativeSearchQueryBuilder完成：

 @Test
    public void testSort(){
        // 构建查询条件
        NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder();
        // 添加基本的分词查询
        queryBuilder.withQuery(QueryBuilders.termQuery("category", "手机"));

        // 排序
        queryBuilder.withSort(SortBuilders.fieldSort("price").order(SortOrder.DESC));

        // 执行搜索，获取结果
        Page<Item> items = this.itemRepository.search(queryBuilder.build());
        // 打印总条数
        System.out.println(items.getTotalElements());
        items.forEach(System.out::println);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

结果：

4、聚合

4.1、聚合为桶

桶就是分组，比如这里我们按照品牌brand进行分组：

    @Test
    public void testAgg(){
        NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder();
        // 不查询任何结果
        queryBuilder.withSourceFilter(new FetchSourceFilter(new String[]{""}, null));
        // 1、添加一个新的聚合，聚合类型为terms，聚合名称为brands
        ，聚合字段为brand
        queryBuilder.addAggregation(
                AggregationBuilders.terms("brands").field("brand"));
        // 2、查询,需要把结果强转为AggregatedPage类型
        AggregatedPage<Item> aggPage = (AggregatedPage<Item>) this.itemRepository.search(queryBuilder.build());
        // 3、解析
        // 3.1、从结果中取出名为brands的那个聚合，
        // 因为是利用String类型字段来进行的term聚合，所以结果要强转为StringTerm类型
        StringTerms agg = (StringTerms) aggPage.getAggregation("brands");
        // 3.2、获取桶
        List<StringTerms.Bucket> buckets = agg.getBuckets();
        // 3.3、遍历
        for (StringTerms.Bucket bucket : buckets) {
            // 3.4、获取桶中的key，即品牌名称
            System.out.println(bucket.getKeyAsString());
            // 3.5、获取桶中的文档数量
            System.out.println(bucket.getDocCount());
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

结果：

AggregationBuilders：聚合的构建工厂类。所有聚合都由这个类来构建，看看他的静态方法：

AggregatedPage：聚合查询的结果类。它是Page的子接口；
AggregatedPage在Page功能的基础上，拓展了与聚合相关的功能，它其实就是对聚合结果的一种封装，大家可以对照聚合结果的JSON结构来看。

而返回的结果都是Aggregation类型对象，不过根据字段类型不同，又有不同的子类表示：

我们看下页面的查询的JSON结果与Java类的对照关系：

4.2、嵌套聚合，求平均值

@Test
public void testSubAgg(){
    NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder();
    // 不查询任何结果
    queryBuilder.withSourceFilter(new FetchSourceFilter(new String[]{""}, null));
    // 1、添加一个新的聚合，聚合类型为terms，聚合名称为brands，聚合字段为brand
    queryBuilder.addAggregation(
        AggregationBuilders.terms("brands").field("brand")
        .subAggregation(AggregationBuilders.avg("priceAvg").field("price")) // 在品牌聚合桶内进行嵌套聚合，求平均值
    );
    // 2、查询,需要把结果强转为AggregatedPage类型
    AggregatedPage<Item> aggPage = (AggregatedPage<Item>) this.itemRepository.search(queryBuilder.build());
    // 3、解析
    // 3.1、从结果中取出名为brands的那个聚合，
    // 因为是利用String类型字段来进行的term聚合，所以结果要强转为StringTerm类型
    StringTerms agg = (StringTerms) aggPage.getAggregation("brands");
    // 3.2、获取桶
    List<StringTerms.Bucket> buckets = agg.getBuckets();
    // 3.3、遍历
    for (StringTerms.Bucket bucket : buckets) {
        // 3.4、获取桶中的key，即品牌名称  3.5、获取桶中的文档数量
        System.out.println(bucket.getKeyAsString() + "，共" + bucket.getDocCount() + "台");

        // 3.6.获取子聚合结果：
        InternalAvg avg = (InternalAvg) bucket.getAggregations().asMap().get("priceAvg");
        System.out.println("平均售价：" + avg.getValue());
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

结果：

五、elasticsearch与MySQL数据同步

5.1 Logstash

5.1.1什么是Logstash

Logstash是一款轻量级的日志搜集处理框架，可以方便的把分散的、多样化的日志搜集起来，并进行自定义的处理，然后传输到指定的位置，比如某个服务器或者文件。

5.1.2 Logstash安装与测试

官网下载地址：https://www.elastic.co/cn/downloads/logstash

下载后解压，进入bin目录，执行如下命令

logstash ‐e 'input { stdin { } } output { stdout {} }'
1

控制台输入字符，随后就有日志输出

• stdin，表示输入流，指从键盘输入
• stdout，表示输出流，指从显示器输出

命令行参数:

• -e 执行
• –config（config前边是两个横杠，这里显示不出来） 或 -f 配置文件，后跟参数类型可以是一个字符串的配置或全路径文件名或全路径
  如：/etc/logstash.d/，logstash会自动读取/etc/logstash.d/目录下所有*.conf 的文
  本文件，然后在自己内存里拼接成一个完整的大配置文件再去执行

6.2 MySQL数据导入Elasticsearch

1. 在logstash安装目录下创建文件夹mysqletc （名称随意）
2. 文件夹下创建mysql.conf （名称随意） ，内容如下：

input {
	jdbc {
		# mysql jdbc connection string to our backup databse 后面的test对应mysql中的test数据库
		jdbc_connection_string =>
		"jdbc:mysql://127.0.0.1:3306/tensquare_article?characterEncoding=UTF8"
		# the user we wish to excute our statement as
		jdbc_user => "root"
		jdbc_password => "123456"
		# the path to our downloaded jdbc driver
		jdbc_driver_library => "D:/logstash‐5.6.8/mysqletc/mysql‐
		connector‐java‐5.1.46.jar"
		# the name of the driver class for mysql
		jdbc_driver_class => "com.mysql.jdbc.Driver"
		jdbc_paging_enabled => "true"
		jdbc_page_size => "50000"
		#以下对应着要执行的sql的绝对路径。
		statement => "select id,title,content from tb_article"
		#定时字段 各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
		schedule => "* * * * *"
	}
} 

output {
	elasticsearch {
			#ESIP地址与端口
			hosts => "localhost:9200"
			#ES索引名称（自己定义的）
			index => "tensquare"
			#自增ID编号
			document_id => "%{id}"
			document_type => "article"
		} 
		stdout {
			#以JSON格式输出
			codec => json_lines
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

3. 将mysql驱动包mysql-connector-java-5.1.46.jar拷贝至/logstash/mysqletc/ 下 。

4. 命令行下执行

logstash ‐f ../mysqletc/mysql.conf
1

观察控制台输出，每间隔1分钟就执行一次sql查询。

再次刷新elasticsearch-head的数据显示，看是否也更新了数据。