Elasticsearch Java API 的使用(9)— Bluk的批量原理 & Bulk大数据量的批量上传示例「建议收藏」

Elasticsearch Java API 的使用(9)— Bluk的批量原理 & Bulk大数据量的批量上传示例「建议收藏」Elasticsearch对多个文档进行索引的简便方法

Elasticsearch

Elasticsearch对多个文档进行索引的简便方法。Bulk api的支持可以实现一次请求执行批量的添加、删除、更新等操作。Bulk操作使用的是UDP协议,UDP无法确保与ElasticSearch服务器通信时不丢失数据。
将数据累计到一定数量进行批量上传,大大加快数据上传速度。

下面代码中,在 “具体操作添加到批量请求中” 前可以是新增、删除、更新等操作。

public class Bulk implements WriteRequest{ 
   

    public static void main(String[] args) { 
   
        String globalname = "";
        String valuep = "";
        String time = "";
        //数据条数记步
        int count = 0;
        //数据上传条数设定
        int BULK_SIZE = 100;
        try{ 
   
            /*创建客户端*/
            //client startup
            //设置集群名称
            Settings settings = Settings.builder()
                    .put("cluster.name", "elsearch")
                    .put("client.transport.sniff", true)
                    .build();
            //创建client
            TransportClient client = new 
                    PreBuiltTransportClient(settings)
                    .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("10.122.4.79"),9300));

            for(int i = 0; i<10; i++){ 
   
                BulkRequestBuilder bulkRequest = client.prepareBulk();
                bulkRequest.setRefreshPolicy(RefreshPolicy.IMMEDIATE);
                XContentBuilder builder=XContentFactory.jsonBuilder()
                                    .startObject();
                //TODO: 设置属性和值
                builder.field("pointid",globalname);
                builder.field("pointvalue",valuep);
                builder.field("inputtime",new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(time));
                builder.endObject();
                
                // 添加索引数据 
                IndexRequestBuilder requestBuilder = client
                        .prepareIndex("pointdata", "pointdata")
                        .setSource(builder);
                
                // 具体操作添加到批量请求中
                bulkRequest.add(requestBuilder);
                count++;
                if(count % BULK_SIZE == 0){ 
   
                    BulkResponse bulkResponse = bulkRequest
                            .execute().actionGet();
                    if (bulkResponse.hasFailures()) { 
   
                        // LOGGER.error("导入索引数据失败: "+"pointdata");
                        // LOGGER.error("导入索引数据失败: "+bulkResponse.buildFailureMessage());
                        System.out.println("导入索引数据失败");
                    }

                    bulkRequest = client.prepareBulk();
                    bulkRequest.setRefreshPolicy(RefreshPolicy.IMMEDIATE);
                    count = 0;
                }
            }
        }catch(Exception e){ 
   
            e.printStackTrace();
        }


    }

    @Override
    public void readFrom(StreamInput arg0) throws IOException { 
   
        // TODO Auto-generated method stub

    }

    @Override
    public void writeTo(StreamOutput arg0) throws IOException { 
   
        // TODO Auto-generated method stub

    }

    @Override
    public RefreshPolicy getRefreshPolicy() { 
   
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public WriteRequest setRefreshPolicy(RefreshPolicy arg0) { 
   
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public ActionRequestValidationException validate() { 
   
        // TODO Auto-generated method stub
        return null;
    }

}

今天的文章Elasticsearch Java API 的使用(9)— Bluk的批量原理 & Bulk大数据量的批量上传示例「建议收藏」分享到此就结束了,感谢您的阅读。

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。
如需转载请保留出处:https://bianchenghao.cn/88770.html

(0)
编程小号编程小号

相关推荐

发表回复

您的电子邮箱地址不会被公开。 必填项已用 * 标注