ElasticsearchRecordWriter.java

package com.yoslab.process.writer;

import java.util.List;

import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.transport.InetSocketTransportAddress;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

import com.yoslab.exception.CommonException;
import com.yoslab.model.ItemRecord;
import com.yoslab.model.MapItemRecord;
import com.yoslab.process.ItemWriter;

public class ElasticsearchRecordWriter implements ItemWriter<ItemRecord> {
	
	Client client = null;
	
	@Override
	public void write(List<ItemRecord> items) throws CommonException {
	        if(items.size() == 0) return;
	        
	    	BulkRequestBuilder bulkRequest = client.prepareBulk();
	        for(ItemRecord item : items){
	        	MapItemRecord m = (MapItemRecord)item;
	        	ObjectMapper mapper = new ObjectMapper();
	        	String json = null;
	        	try {
				json = mapper.writeValueAsString(m);
			} catch (JsonProcessingException e) {
		        	throw new CommonException(null,"json変換時にエラーが発生しました。",e);
			}
	        	
	        	//TODO need to get field name for id.
	        	String id = (String)m.getValue("id");
	        	
	        	//TODO set index name & type name
	//			IndexResponse response = client
	//					.prepareIndex("posts", "article", id).setSource(json)
	//					.execute().actionGet();
	        	//using bulk request.
	        	bulkRequest.add(client.prepareIndex("posts", "article", id).setSource(json));
	        }
	        
	        BulkResponse bulkResponse = bulkRequest.execute().actionGet();
	        if (bulkResponse.hasFailures()) {
	            throw new CommonException(null,bulkResponse.buildFailureMessage());
	        }
	}
	
	@Override
	public void postProcess() throws CommonException {
		//NOOP
	}

	@Override
	public void commit() throws CommonException {
		//NOOP
	}

	@Override
	public void close() {
		client.close();
	}
	
	@Override
	public void init() throws CommonException {
		client = new TransportClient()
				.addTransportAddress(new InetSocketTransportAddress(
						"localhost", 9300));
	}
}

depends on following jars.

• elasticsearch-1.3.2.jar
• lucene-core-4.9.0.jar
• jackson-databind-2.4.2.jar
• jackson-core-2.4.2.jar
• jackson-annotations-2.4.2.jar

You cannot connect ES over HTTP Proxy. Because Elaticsearch and Client API is talking with TCP.