ES —BulkProcessor，批量创建索引

末蓝、 2022-05-21 11:25 578阅读 0赞

场景：单条循环，数量大效率低，执行半个小时cpu就超标了……

解决方案：BulkProcessor批量插入；[官网地址][Link 1]

网上找了很多，都一堆代码，看得莫名其妙\*\*\*\*\*明明很简单的几句而已

关键是：bulkProcessor.add()方法；

public void bulkProcess() {
    
            BulkProcessor bulkProcessor = BulkProcessor.builder(ESClientUtils.client,
                    new BulkProcessor.Listener() {
                        @Override
                        public void beforeBulk(long executionId,
                                               BulkRequest request) {
    
                        }
    
                        @Override
                        public void afterBulk(long executionId, BulkRequest request,  BulkResponse response) {
    
                        }
                        @Override
                        public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                            System.out.println("error：" + failure.getMessage() + " \ncause：" +failure.getCause());
                        }
                    })
                    .setBulkActions(10000)
                    .setBulkSize(new ByteSizeValue(20, ByteSizeUnit.MB))
                    .setFlushInterval(TimeValue.timeValueSeconds(5))
                    .setConcurrentRequests(1)
                    .build();
    
    
            for(int i=0;i<1000;i++){
                bulkProcessor.add(new IndexRequest("index", "type","自定义id").source("索引数据-map等格式"));
            }
            bulkProcessor.close();
        }

说明：ESClientUtils.client用于获取ES的client，这里就不多累赘

旧版如下：单条数据插入

IndexResponse response = ESClientUtils.client.prepareIndex(config.getES_index(), config.getES_type()).setId(proId).setSource(data).execute().actionGet();

[Link 1]: https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/java-docs-bulk-processor.html