Spring Boot + Elasticsearch 实现索引批量写入

在使用Eleasticsearch进行索引维护的过程中,如果你的应用场景需要频繁的大批量的索引写入,再使用上篇中提到的维护方法的话显然效率是低下的,此时推荐使用bulkIndex来提升效率。批写入数据块的大小取决于你的数据集及集群的配置。

下面我们以Spring Boot结合Elasticsearch创建一个示例项目,从基本的pom配置开始

  1. <dependency>
  2.    <groupId>com.google.code.gson</groupId>
  3.    <artifactId>gson</artifactId>
  4.    <version>1.4</version>
  5. </dependency>
  6. <dependency>
  7.    <groupId>org.springframework.boot</groupId>
  8.    <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
  9. </dependency>

application.properties配置

  1. #elasticsearch config
  2. spring.data.elasticsearch.cluster-name:elasticsearch
  3. spring.data.elasticsearch.cluster-nodes:192.168.1.105:9300
  4. #application config
  5. server.port=8080
  6. spring.application.name=esp-app

我们需要定义域的实体和一个Spring data的基本的CRUD支持库类。用id注释定义标识符字段,如果你没有指定ID字段,Elasticsearch不能索引你的文件。同时需要指定索引名称类型,@Document注解也有助于我们设置分片和副本数量。

  1. @Data
  2. @Document(indexName = "carIndex", type = "carType", shards = 1, replicas = 0)
  3. public class Car implements Serializable {
  4.    /**
  5.     * serialVersionUID:
  6.     * @since JDK 1.6
  7.     */
  8.    private static final long serialVersionUID = 1L;
  9.    @Id
  10.    private Long id;
  11.    private String brand;
  12.    private String model;
  13.    private BigDecimal amount;
  14.    public Car(Long id, String brand, String model, BigDecimal amount) {
  15.        this.id = id;
  16.        this.brand = brand;
  17.        this.model = model;
  18.        this.amount = amount;
  19.    }
  20. }

接着定义一个IndexService并使用bulk请求来处理索引,操作前首先要判断索引是否存在,以免出现异常。为了更好的掌握Java API,这里采用了不同于上篇中ElasticSearchRepository的ElasticSearchTemplate工具集,相对来讲功能更加丰富。

  1. @Service
  2. public class IndexerService {
  3.    private static final String CAR_INDEX_NAME = "car_index";
  4.    private static final String CAR_INDEX_TYPE = "car_type";
  5.    @Autowired
  6.    ElasticsearchTemplate elasticsearchTemplate;
  7.    public long bulkIndex() throws Exception {
  8.        int counter = 0;
  9.        try {
  10.            //判断索引是否存在
  11.            if (!elasticsearchTemplate.indexExists(CAR_INDEX_NAME)) {
  12.                elasticsearchTemplate.createIndex(CAR_INDEX_NAME);
  13.            }
  14.            Gson gson = new Gson();
  15.            List<IndexQuery> queries = new ArrayList<IndexQuery>();
  16.            List<Car> cars = assembleTestData();
  17.            for (Car car : cars) {
  18.                IndexQuery indexQuery = new IndexQuery();
  19.                indexQuery.setId(car.getId().toString());
  20.                indexQuery.setSource(gson.toJson(car));
  21.                indexQuery.setIndexName(CAR_INDEX_NAME);
  22.                indexQuery.setType(CAR_INDEX_TYPE);
  23.                queries.add(indexQuery);
  24.                //分批提交索引
  25.                if (counter % 500 == 0) {
  26.                    elasticsearchTemplate.bulkIndex(queries);
  27.                    queries.clear();
  28.                    System.out.println("bulkIndex counter : " + counter);
  29.                }
  30.                counter++;
  31.            }
  32.            //不足批的索引最后不要忘记提交
  33.            if (queries.size() > 0) {
  34.                elasticsearchTemplate.bulkIndex(queries);
  35.            }
  36.            elasticsearchTemplate.refresh(CAR_INDEX_NAME);
  37.            System.out.println("bulkIndex completed.");
  38.        } catch (Exception e) {
  39.            System.out.println("IndexerService.bulkIndex e;" + e.getMessage());
  40.            throw e;
  41.        }
  42.        return -1;
  43.    }
  44.    private List<Car> assembleTestData() {
  45.        List<Car> cars = new ArrayList<Car>();
  46.        //随机生成10000个索引,以便下一次批量写入
  47.        for (int i = 0; i < 10000; i++) {
  48.            cars.add(new Car(RandomUtils.nextLong(1, 11111), RandomStringUtils.randomAscii(20), RandomStringUtils.randomAlphabetic(15), BigDecimal.valueOf(78000)));
  49.        }
  50.        return cars;
  51.    }
  52. }

再下面的工作就比较简单了,可以编写一个RestController接受请求来测试或者CommandLineRunner,在系统启动时就加载上面的方法。

  1. @SpringBootApplication
  2. @RestController
  3. public class ESPApplicatoin {
  4.    public static void main(String[] args) {
  5.        SpringApplication.run(ESPApplicatoin.class, args);
  6.    }
  7.    @Autowired
  8.    IndexerService indexService;
  9.    @RequestMapping(value = "bulkIndex",method = RequestMethod.POST)
  10.    public void bulkIndex(){
  11.        try {
  12.            indexService.bulkIndex();
  13.        } catch (Exception e) {
  14.            e.printStackTrace();
  15.        }
  16.    }
  17. }

CommandLineRunner方法类:

  1. @Component
  2. public class AppLoader implements CommandLineRunner {
  3.    @Autowired
  4.    IndexerService indexerService;
  5.    @Override
  6.    public void run(String... strings) throws Exception {
  7.        indexerService.bulkIndex();
  8.    }
  9. }

结束后,就可在通过地址http://localhost:9200/car_index/_search/来查看索引到底有无生效。注:要特别关注版本的兼容问题,如果用Es
5+的话,显然不能采用Spring Data Elasticsearch的方式。

Spring Boot

Version (x)

Spring Data Elasticsearch Version (y) Elasticsearch Version (z)
x <= 1.3.5 y <= 1.3.4 z <= 1.7.2*
x >= 1.4.x 2.0.0 <=y < 5.0.0** 2.0.0 <= z < 5.0.0**

(*) - require manual change in your project pom file (solution 2.)

(**) - Next big ES release with breaking changes

>>>案例地址:https://github.com/backkoms/spring-boot-elasticsearch

扩展阅读:

Spring
Boot + Elasticsearch 实现索引的日常维护

基于SpringCloud的Microservices架构实战案例-序篇

Nginx+Lua+MySQL/Redis实现高性能动态网页展现

Nginx+Lua+Redis实现高性能缓存数据读取

Spring Boot + Elasticsearch 实现索引批量写入

上一篇:Spring Boot + Elasticsearch 实现索引的日常维护


下一篇:hdu 4641 K-string SAM的O(n^2)算法 以及 SAM+并查集优化