Bulk API: To perform bulk operations in elasticsearch using java

Elasticsearch: Java: Bulk API: To perform bulk operations

Bulk api allows us to perform multiple operations at a time. One way to create bulk request is using BulkRequestBuilder class.

String _index = "organization";
String _type = "employee";
String _id1 = 1;
String _id2 = 2;

BulkRequestBuilder bulkRequest = client.prepareBulk();

// either use client#prepare, or use Requests# to directly build index/delete requests
bulkRequest.add(client.prepareIndex(_index, _type, _id1)
        .setSource(jsonBuilder()
                    .startObject()
                        .field("user", "Krishna")
                        .field("postDate", new Date())
                        .field("message", "trying out Elasticsearch")
                    .endObject()
                  )
        );

bulkRequest.add(client.prepareIndex(_index, _type, _id2)
        .setSource(jsonBuilder()
                    .startObject()
                        .field("user", "PYR")
                        .field("postDate", new Date())
                        .field("message", "another post")
                    .endObject()
                  )
        );

BulkResponse bulkResponse = bulkRequest.execute().actionGet();
if (bulkResponse.hasFailures()) {
    // process failures by iterating through each bulk response item
}

Using BulkProcessor
The BulkProcessor class offers a simple interface to flush bulk operations automatically based on the number or size of requests, or after a given period.

import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;

String _index = "organization";
String _type = "employee";
String _id1 = 1;
String _id2 = 2;

BulkProcessor bulkProcessor = BulkProcessor.builder(
        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) { ... }
        })
        .setBulkActions(10000)
        .setBulkSize(new ByteSizeValue(1, ByteSizeUnit.GB))
        .setFlushInterval(TimeValue.timeValueSeconds(5))
        .setConcurrentRequests(1)
        .build();
     
bulkProcessor.add(new IndexRequest(_index, _type, _id1).source(/* your doc here */));
bulkProcessor.add(new DeleteRequest(_index, _type, _id2));


@Override
public void beforeBulk(long executionId,BulkRequest request) { ... }
This method is called just before bulk is executed.

@Override
public void afterBulk(long executionId, BulkRequest request,BulkResponse response) { ... }
This method is called after bulk execution. You can check for some failing requests with response.hasFailures().

@Override
public void afterBulk(long executionId,BulkRequest request,Throwable failure) { ... }
This method is called when the bulk failed and raised a Throwable

setBulkActions(10000)
We want to execute the bulk every 10000 requests

setBulkSize(new ByteSizeValue(1, ByteSizeUnit.GB))
We want to flush the bulk every 1gb

setFlushInterval(TimeValue.timeValueSeconds(5))
We want to flush the bulk every 5 seconds whatever the number of requests

.setConcurrentRequests(1)
Set the number of concurrent requests. A value of 0 means that only a single request will be allowed to be executed. A value of 1 means 1 concurrent request is allowed to be executed while accumulating new bulk requests.

By default, BulkProcessor:
    sets bulkActions to 1000
    sets bulkSize to 5mb
    does not set flushInterval
    sets concurrentRequests to 1

Following application demonstrates simple Bulk utility.

Step 1:Define simple model class Employee.

package com.self_learn.model;

import java.util.ArrayList;
import java.util.List;

import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;

@EqualsAndHashCode()
@ToString
public class Employee {
@Getter @Setter private String age;
@Getter @Setter private String firstName;
@Getter @Setter private String lastName;
@Getter @Setter private List<String> hobbies = new ArrayList<>();
}


Step 2: Define Utility class to get Client instance.

package com.self_learn.util;

import static com.self_learn.util.IPUtil.isValidHosts;
import static com.self_learn.util.IPUtil.isValidPorts;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;

import com.google.common.base.Preconditions;

public class TransportClientUtil {
private static Map<Map<String, Integer>, Client> localMap = new HashMap<>();

/**
* Take machine name and port addresses as map and return transport client.
* Key is host name, value is port number
*
* @throws UnknownHostException
*/
public static Client getTransportClient(String clusterName,
Map<String, Integer> map) throws UnknownHostException {

Preconditions.checkNotNull(clusterName,
"clusterName shouldn't be empty");
Preconditions.checkNotNull(map, "Map shouldn't be empty");

if (localMap.containsKey(map))
return localMap.get(map);

Preconditions.checkState(isValidHostPorts(map),
"Map contains invalid host (or) port");

Settings settings = ImmutableSettings.settingsBuilder()
.put("cluster.name", clusterName)
.put("client.transport.sniff", true).build();

TransportClient client = new TransportClient(settings);

InetSocketTransportAddress addresses[] = getInetSocketTransportAddresses(map);
client.addTransportAddresses(addresses);
localMap.put(map, client);
return client;
}

/**
* @param map
* @return true, if all the entries in map are valid host, ports. Else
*         false.
*/
private static boolean isValidHostPorts(Map<String, Integer> map) {
Set<String> hostNames = map.keySet();
Set<Integer> ports = new HashSet<>(map.values());

if (!isValidHosts(hostNames.toArray(new String[hostNames.size()])))
return false;

if (!isValidPorts(ports.toArray(new Integer[ports.size()])))
return false;

return true;
}

private static InetSocketTransportAddress[] getInetSocketTransportAddresses(
Map<String, Integer> map) throws UnknownHostException {
InetSocketTransportAddress addresses[] = new InetSocketTransportAddress[map
.size()];
int count = 0;

Set<String> keys = map.keySet();
for (String key : keys) {
InetAddress addr = InetAddress.getByName(key);
InetSocketTransportAddress address = new InetSocketTransportAddress(
addr, map.get(key));
addresses[count] = address;
}

return addresses;
}

/**
* Get transport client for localhost.
*
* @param clusterName
* @param port
* @return
* @throws UnknownHostException
*/
public static Client getLocalTransportClient(String clusterName, int port)
throws UnknownHostException {
Settings settings = ImmutableSettings.settingsBuilder()
.put("cluster.name", clusterName)
.put("client.transport.sniff", true).build();

TransportClient client = new TransportClient(settings);
InetAddress addr = InetAddress.getByName("127.0.0.1");
InetSocketTransportAddress address = new InetSocketTransportAddress(
addr, port);

client.addTransportAddress(address);
return client;
}

}


Following is the IPUtil class to validate hostnames, ports.

package com.self_learn.util;

import org.apache.commons.validator.routines.InetAddressValidator;

import com.google.common.base.Preconditions;

/**
 * Validate IPaddresses, ports
 *
 * @author harikrishna_gurram
 */
public class IPUtil {
private static InetAddressValidator inetAddressValidator = InetAddressValidator
.getInstance();

/**
* @param ipAddress
* @return true if ip address is valid, else false
*/
public static boolean isValidIPAddress(String ipAddress) {
Preconditions.checkNotNull(ipAddress, "IP address should not be null");
return inetAddressValidator.isValid(ipAddress);
}

/**
* @param port
*            : Port number
* @return true if port number is valid, else false
*/
public static boolean isValidPort(int port) {
if (port > 0 && port < 65536)
return true;
return false;
}

/**
* @param hostNames
* @return true if all the elements of array represents valid hosnames, else
*         false.
*/
public static boolean isValidHosts(String[] hostNames) {
Preconditions.checkNotNull(hostNames, "Host names shouldn't be empty");
for (String hostName : hostNames) {
if (!isValidIPAddress(hostName)) {
return false;
}
}
return true;
}

/**
*
* @param ports
* @return true if all the elements of array represents valid ports, else
*         false.
*/
public static boolean isValidPorts(Integer[] ports) {
Preconditions.checkNotNull(ports, "ports shouldn't be empty");
for (int port : ports) {
if (!isValidPort(port)) {
return false;
}
}
return true;
}

}


Step 3: BulkRequestType class defines constants to identify type of the request like Index, update, upsert update, and delete.

package com.self_learn.constants;

/**
 * Specifies the type of BulkRequest
 *
 * @author harikrishna_gurram
 */
public enum BulkRequestType {
INDEX, UPDATE, UPSERT_UPDATE, DELETE
}


Step 4: Define BulkModelObject. Each BulkObject instance represents a request like INDEX, UPDATE, UPSERT_UPDATE,DELETE.

package com.self_learn.model;

import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;

import com.self_learn.constants.BulkRequestType;

@EqualsAndHashCode()
@ToString
public class BulkObject {

@Getter
@Setter
private String _index;
@Getter
@Setter
private String _type;
@Getter
@Setter
private String _id;
@Getter
@Setter
private String _source;
@Getter
@Setter
private String _upsertSource;
@Getter
@Setter
private BulkRequestType type;

public BulkObject(String _index, String _type, String _id,
BulkRequestType type) {
this._index = _index;
this._type = _type;
this._id = _id;
this.type = type;
}

public BulkObject(String _index, String _type, String _id,
BulkRequestType type, String _source) {
this(_index, _type, _id, type);
this._source = _source;
}

public BulkObject(String _index, String _type, String _id,
BulkRequestType type, String _source, String _upsertSource) {
this(_index, _type, _id, type, _source);
this._upsertSource = _upsertSource;
}
}


Step 5: Define BulkUtil class, which takes set of BulkObjects and process them.

package com.self_learn.util;

import java.util.Set;

import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.base.Preconditions;

import com.self_learn.constants.BulkRequestType;
import com.self_learn.model.BulkObject;

/**
 * Utility class to execute bulk requests.
 *
 * @author harikrishna_gurram
 */
public class BulkUtil {

private static BulkObject getBulkObject(String _index, String _type,
String _id, BulkRequestType type) {
return new BulkObject(_index, _type, _id, type);
}

private static BulkObject getBulkObject(String _index, String _type,
String _id, BulkRequestType type, String _source) {
return new BulkObject(_index, _type, _id, type, _source);
}

private static BulkObject getBulkObject(String _index, String _type,
String _id, BulkRequestType type, String _source,
String _upsertSource) {
return new BulkObject(_index, _type, _id, type, _source, _upsertSource);
}

public static BulkObject getIndexObject(String _index, String _type,
String _id, String _source) {
return getBulkObject(_index, _type, _id, BulkRequestType.INDEX, _source);
}

public static BulkObject getDeleteObject(String _index, String _type,
String _id) {
return getBulkObject(_index, _type, _id, BulkRequestType.DELETE);
}

public static BulkObject getUpdateObject(String _index, String _type,
String _id, String src) {
return getBulkObject(_index, _type, _id, BulkRequestType.UPDATE, src);
}

public static BulkObject getUpsertUpdateObject(String _index, String _type,
String _id, String _source, String _upsertSource) {
return getBulkObject(_index, _type, _id, BulkRequestType.UPSERT_UPDATE,
_source, _upsertSource);
}

public static BulkObject getIndexObject(String _index, String _type,
String _id, Object _source) {
return getIndexObject(_index, _type, _id, JSONUtil.getJson(_source));
}

public static BulkObject getUpdateObject(String _index, String _type,
String _id, Object _source) {
return getUpdateObject(_index, _type, _id, JSONUtil.getJson(_source));
}

public static BulkObject getUpsertUpdateObject(String _index, String _type,
String _id, Object _source, Object _upsertSource) {
return getUpsertUpdateObject(_index, _type, _id,
JSONUtil.getJson(_source), JSONUtil.getJson(_upsertSource));
}

private static boolean isNull(Object obj) {
return (obj == null);
}

private static boolean isNotNull(Object obj) {
return !isNull(obj);
}

/**
* Validate BulkObject based on Request type.
*
* If Request type
*
* a. is @{link BulkRequestType.INDEX} then BulkObject must has _index,
* _type, _id, _source.
*
* b. is @{link BulkRequestType.UPDATE} then BulkObject must has _index,
* _type, _id, _source.
*
* c. is @{link BulkRequestType.UPSERT_UPDATE} then BulkObject must has
* _index, _type, _id, _source, _upsertSource.
*
* d. is @{link BulkRequestType.DELETE} then BulkObject must has _index,
* _type, _id.
*
* @param type
*            Represents Request type, it can be INDEX, UPDATE,
*            UPSERT_UPDATE, DELETE.
*
* @param obj
* @return
*/
private static boolean isValidBulkObject(BulkObject obj) {
if (obj == null)
return false;

String _index = obj.get_index();
String _type = obj.get_type();
String _id = obj.get_id();
String _source = obj.get_source();
String _upsertSource = obj.get_upsertSource();
BulkRequestType type = obj.getType();

switch (type) {
case INDEX:
if (isNotNull(_index) && isNotNull(_type) && isNotNull(_id)
&& isNotNull(_source))
return true;
break;
case UPDATE:
if (isNotNull(_index) && isNotNull(_type) && isNotNull(_id)
&& isNotNull(_source))
return true;
break;
case UPSERT_UPDATE:
if (isNotNull(_index) && isNotNull(_type) && isNotNull(_id)
&& isNotNull(_source) && isNotNull(_upsertSource))
return true;
break;
case DELETE:
if (isNotNull(_index) && isNotNull(_type) && isNotNull(_id))
return true;
break;
}
return false;
}

/**
* method takes Map of requests and execute bulk requests. Key of map
* represents type of request.
*
* @{link BulkRequestType.INDEX} represents Index request
* @{link BulkRequestType.UPDATE} represents update request
* @{link BulkRequestType.UPSERT_UPDATE} represents UPSERT_UPDATE request
* @{link BulkRequestType.DELETE} represents delete request
*
*        If BulkObject is invalid, then the request is simply ignored.
*
* @param bulkRequest
* @return
*/
public static BulkResponse execute(Client client, Set<BulkObject> bulkReq) {
Preconditions.checkNotNull(client, "client shouldn't be null");
Preconditions.checkNotNull(bulkReq, "bulkReq shouldn't be null");

BulkRequestBuilder bulkRequest = client.prepareBulk();

for (BulkObject obj : bulkReq) {
if (!isValidBulkObject(obj))
continue;

String _index = obj.get_index();
String _type = obj.get_type();
String _id = obj.get_id();
String _source = obj.get_source();
String _upsertSource = obj.get_upsertSource();

BulkRequestType type = obj.getType();

switch (type) {
case INDEX:
bulkRequest.add(new IndexRequest(_index, _type, _id)
.source(_source));
break;
case UPDATE:
bulkRequest.add(new UpdateRequest(_index, _type, _id)
.doc(_source));
break;
case UPSERT_UPDATE:
bulkRequest.add(new UpdateRequest(_index, _type, _id).doc(
_source).upsert(_upsertSource));
break;
case DELETE:
bulkRequest.add(new DeleteRequest(_index, _type, _id));
break;

default:
System.out.println("Invalid Request");
}
}

return bulkRequest.execute().actionGet();
}
}


Step 6: Define ResponseUtil class, used to return the response of query in json format.

package com.self_learn.util;

import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.get.GetResponse;

import com.google.common.base.Preconditions;

/**
 * Utility class to return response in string format.
 *
 * @author harikrishna_gurram
 */
public class ResponseUtil {

/**
* Returns source of the result as string.
*
* @param response
* @return
*/
public static String getSource(GetResponse response, boolean pretty) {
Preconditions.checkNotNull(response, "response shouldn't be null");
return response.getSourceAsString();
}

/**
* @param response
* @return string representation of {@link BulkResponse}
*/
public static String getResponseInfo(Object response, boolean pretty) {
if (pretty)
return JSONUtil.getPrettyJson(response);

return JSONUtil.getJson(response);
}
}


Step 8: Main.java demonstrate how to use BulkUtil class.

package com.self_learn.test;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;

import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.Client;

import com.self_learn.model.BulkObject;
import com.self_learn.model.Employee;
import com.self_learn.util.BulkUtil;
import com.self_learn.util.ResponseUtil;
import com.self_learn.util.TransportClientUtil;

public class Main {
private static String clusterName = "my_cluster_1";
private static String _index = "organization";
private static String _type = "employee";

public static void main(String args[]) throws IOException,
InterruptedException, ExecutionException {
/* Get client instance for cluster */
Client client = TransportClientUtil.getLocalTransportClient(
clusterName, 9300);

Set<BulkObject> objects = new HashSet<>();

/* First 5 index requests */
for (int i = 1; i <= 5; i++) {
Employee emp = new Employee();
emp.setAge("" + i);
emp.setFirstName("firstName " + i);
emp.setLastName("lastName " + i);
emp.setHobbies(Arrays.asList("hobbies " + i));
BulkObject obj = BulkUtil
.getIndexObject(_index, _type, "" + i, emp);
objects.add(obj);
}

BulkResponse response = BulkUtil.execute(client, objects);
System.out.println(ResponseUtil.getResponseInfo(response, true));

/* Delete documents 1 and 4 */
objects = new HashSet<>();
objects.add(BulkUtil.getDeleteObject(_index, _type, "1"));
objects.add(BulkUtil.getDeleteObject(_index, _type, "4"));
response = BulkUtil.execute(client, objects);
System.out.println(ResponseUtil.getResponseInfo(response, true));

/* Update document2 and index document 6 */
objects = new HashSet<>();
Map<String, Object> doc = new HashMap<>();
doc.put("firstName", "Krishna");
doc.put("age", "26");
objects.add(BulkUtil.getUpdateObject(_index, _type, "2", doc));

Employee emp = new Employee();
emp.setAge("" + 6);
emp.setFirstName("firstName " + 6);
emp.setLastName("lastName " + 6);
emp.setHobbies(Arrays.asList("hobbies " + 6));
objects.add(BulkUtil.getIndexObject(_index, _type, "6", emp));
response = BulkUtil.execute(client, objects);
System.out.println(ResponseUtil.getResponseInfo(response, true));

client.close();
}
}

Once you ran, Main.java, you will get following output.  
Sep 10, 2015 2:02:14 PM org.elasticsearch.plugins.PluginsService <init>
INFO: [Turner Century] loaded [], sites []
{
  "responses": [
    {
      "id": 0,
      "opType": "index",
      "response": {
        "index": "organization",
        "id": "2",
        "type": "employee",
        "version": 1,
        "created": true
      }
    },
    {
      "id": 1,
      "opType": "index",
      "response": {
        "index": "organization",
        "id": "4",
        "type": "employee",
        "version": 1,
        "created": true
      }
    },
    {
      "id": 2,
      "opType": "index",
      "response": {
        "index": "organization",
        "id": "1",
        "type": "employee",
        "version": 1,
        "created": true
      }
    },
    {
      "id": 3,
      "opType": "index",
      "response": {
        "index": "organization",
        "id": "3",
        "type": "employee",
        "version": 1,
        "created": true
      }
    },
    {
      "id": 4,
      "opType": "index",
      "response": {
        "index": "organization",
        "id": "5",
        "type": "employee",
        "version": 1,
        "created": true
      }
    }
  ],
  "tookInMillis": 2,
  "remoteAddress": {
    "address": {}
  }
}
{
  "responses": [
    {
      "id": 0,
      "opType": "delete",
      "response": {
        "index": "organization",
        "id": "4",
        "type": "employee",
        "version": 2,
        "found": true
      }
    },
    {
      "id": 1,
      "opType": "delete",
      "response": {
        "index": "organization",
        "id": "1",
        "type": "employee",
        "version": 2,
        "found": true
      }
    }
  ],
  "tookInMillis": 1,
  "remoteAddress": {
    "address": {}
  }
}
{
  "responses": [
    {
      "id": 0,
      "opType": "update",
      "response": {
        "index": "organization",
        "id": "2",
        "type": "employee",
        "version": 2,
        "created": false
      }
    },
    {
      "id": 1,
      "opType": "index",
      "response": {
        "index": "organization",
        "id": "6",
        "type": "employee",
        "version": 1,
        "created": true
      }
    }
  ],
  "tookInMillis": 1,
  "remoteAddress": {
    "address": {}
  }
}