ElastaticSearch -- es批量插入数据_es 批量插入

es批量插入数据的示例

    /**
     * 批量插入数据
     * @param client  es客户端
     * 如果没有现成的client， 获取client的代码以及jar包，详情见 https://www.cnblogs.com/expiator/p/15735701.html
     * @param list 对象集合
     * @param index  es索引
     * @param type es索引的类型
     */
    public static void bulkInsert(TransportClient client , List<Person> list, String index, String type) {
        //通过client实例化 BulkRequestBuilder
        BulkRequestBuilder bulkRequest = client.prepareBulk();
        for (Person person : list) {
            String dataStr = JSON.toJSONString(person);
            //最好能保证id唯一，可以拼上时间戳或者主键之类的
            StringBuilder id = new StringBuilder(person.getName()).append("_").append(person.getAge()).append("_").append(person.getAddress());
            //往BulkRequestBuilder中添加数据            
            bulkRequest.add(client.prepareIndex(index, type, id.toString()).setSource(dataStr, XContentType.JSON));
        }
        //执行请求并回调
        BulkResponse bulkItemResponses = bulkRequest.get();
        log.info("es插入是否成功：{}, 插入条数：{}", bulkItemResponses.status().toString(), CollectionUtils.size(list));
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

es批量插入数据

上面的例子使用的是具体的对象集合 List。
对象集合可以转换成 List，再批量插入es，就能统一成通用的方法了。

    /**
     * 批量插入数据
     * @param client  es客户端
     * 如果没有现成的client， 获取client的代码以及jar包，详情见 https://www.cnblogs.com/expiator/p/15735701.html 
     * @param list java对象集合转换而成的json集合
     * @param index  es索引
     * @param type es索引的类型
     * @param idField 插入数据对应的id字段
     */
    public static void bulkInsertJsonList(TransportClient client, List<JSONObject> list, String index, String type, String idField) {
        if (CollectionUtils.isEmpty(list)) {
            return;
        }
        long start = System.currentTimeMillis();
        BulkRequestBuilder bulkRequest = client.prepareBulk();
        for (JSONObject json : list) {
            //根据字段获取对应的id，最好能保证id唯一
            String id = json.getString(idField);
            bulkRequest.add(client.prepareIndex(index, type, id).setSource(json));
        }
        BulkResponse bulkItemResponses = bulkRequest.get();
        log.info("es插入是否成功：{}, 插入条数：{}, 插入耗时时长:{}ms"
                , bulkItemResponses.status().toString(), CollectionUtils.size(list), System.currentTimeMillis() - start);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24