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": {}
}
}
