2025年Elasticsearch Java API 的使用（9）— Bluk的批量原理 & Bulk大数据量的批量上传示例

科技前沿 • 2025-04-03 11:42 • 阅读 24

大家好，我是讯享网，很高兴认识大家。

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

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; } }

讯享网

2025年Elasticsearch Java API 的使用（9）— Bluk的批量原理 & Bulk大数据量的批量上传示例

相关推荐