Demos how to use the ElasticSearch Java API

ElasticSearchDemo v1

/*
 * Short application that illustrates how to use the ElasticSearch Java API.
 * 
 * It is based on the following tutorial:
 * 
 *      http://www.slideshare.net/dadoonet/hands-on-lab-elasticsearch
 *      
 */

package ca.nrc.ElasticSearch;

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Random;

import org.codehaus.jackson.JsonGenerationException;
import org.codehaus.jackson.JsonParseException;
import org.codehaus.jackson.map.JsonMappingException;
import org.codehaus.jackson.map.ObjectMapper;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.deletebyquery.DeleteByQueryResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.node.Node;
import org.elasticsearch.node.NodeBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.client.transport.*;
import org.elasticsearch.common.transport.*;

public class ElasticSearchDemo {

	static ObjectMapper mapper;
	static Client client;
	static String indexName = "meal18";
	static String typeName = "beer";
	static long startTimeMSecs;
	static HashMap<String, Beer> newBeers = new HashMap<String, Beer>();
	

	public static void main(String[] args) throws Exception {
		startTimeMSecs = System.currentTimeMillis();			
		mapper = new ObjectMapper(); // create once, reuse
		
		echo("\n\n=== Creating the ElasticSearch client...");
//		makeClientFromEmbeddedNode(); // Does not raise exceptions, but Marvel/Sense doesn't see the index.
//		makeClientFromNamedClusterNode("elasticsearch"); // Raises an org.elasticsearch.discovery.MasterNotDiscoveredException error
		makeClientFromTransportClient(); // BINGO!!! Initializes in < 1 secs, and Marvel/Sense does see the documents!!!		
		echo("DONE creating the ElasticSearch client...");	

		echo("\n\n=== Creating the "+indexName+" index if it does not exist...");
		createIndex();
		echo("DONE Creating the "+indexName+" index");
		
		echo("\n\n=== Deleting all beers from index "+indexName+"...");
		deleteAllBeers();
		echo("\n\n=== DONE Deleting all beers from index "+indexName+"...");
		
		echo("\n\n=== Indexing two new beer objects one by one...");
		indexTwoNewBeersOneAtATime();
		echo("DONE Indexing two new beer objects one by one...");

		echo("\n\n== Indexing 10 new beer objects as in one batch...");
		index10NewBeersInOneBatch();
		echo("DONE Indexing 10 new beer objects as in one batch...");

		echo("\n\n=== Retrieving all new beer objects...");
		retrieveAndCheckAllNewBeers();
		echo("\n\nDONE Retrieving all new beer objects...");

		
		echo("\n\nDelete all but one of the new beers...");
		deleteAllButOneOfTheNewBeers();
		echo("DONE Delete all but one of the new beers...");
		
		
		echo("\n\n=== Closing the client object...");
		client.close();
		echo("DONE Closing the client object...");

		echo("\n\n=== By now, the index should only contain one beer and its name should be HeinekenNNN where NNN is a number of milliseconds that is less than 1000 lower than "+Long.toString(System.currentTimeMillis())+"."
			 +"\nYou can list the beers in the index by issueing the following query in Marvel/Seense: 'GET /"+indexName+"/_search'");

		echo("\n\n=== Application finished running.");
	}
	
	public static void createIndex() throws IOException {
		
		int numReplicas = 0;
        XContentBuilder settings = XContentFactory.jsonBuilder()
                .startObject()
                        .startObject("index")
                            .field("number_of_shards", 3)
                            .field("number_of_replicas", numReplicas)
                        .endObject();
        try {
        	CreateIndexRequestBuilder irb = client.admin().indices().prepareCreate(indexName);
        	irb.setSettings(settings);
        	irb.execute().actionGet();
        } catch (org.elasticsearch.indices.IndexAlreadyExistsException exc) {
        	echo("WARNING: Could not create index "+indexName+" because it already existed.");
        }
	}
	
	public static void makeClientFromEmbeddedNode() {
		Node node = NodeBuilder.nodeBuilder().node();
		client = node.client();
	}

	public static void makeClientFromNamedClusterNode(String clusterName) {
		Node node = NodeBuilder.nodeBuilder().clusterName(clusterName).client(true).node();
		client = node.client();
	}
	

	public static void makeClientFromTransportClient() {
		client = new TransportClient()
        .addTransportAddress(new InetSocketTransportAddress("localhost", 9300)); // Why 9300 and not 9200???
	}

	public static void deleteAllBeers() {
		echo("Deleting all beers from index "+indexName+"...");
		QueryBuilder qb = QueryBuilders.matchAllQuery();
		SearchResponse sr = client.prepareSearch().setQuery(qb).execute().actionGet();
		long numHits = sr.getHits().getTotalHits();

		DeleteByQueryResponse response = client.prepareDeleteByQuery(indexName)
		        .setQuery(QueryBuilders.termQuery("_type", typeName))
		        .execute()
		        .actionGet();		
		echo("DONE Deleting all beers from index "+indexName+"...");
	}
	
	public static void indexTwoNewBeersOneAtATime() throws JsonGenerationException, JsonMappingException, IOException {
		for (int ii=0; ii < 2; ii++) {
			Beer beer = makeABeer();
			
			echo("Indexing the beer object...");
			String jsonString = mapper.writeValueAsString(beer);
			IndexResponse ir = client.prepareIndex(indexName, typeName).setSource(jsonString)
					.execute().actionGet();
			newBeers.put(ir.getId(), beer);
			echo("DONE Indexing the beer object...");			
		}

	}

	public static void index10NewBeersInOneBatch() throws Exception {

		BulkRequestBuilder brb = client.prepareBulk();
		if (brb == null) {
			throw new Exception("The BulkRequestBuilder instance was null!!!");
			
		}
		for (int ii = 0; ii < 10; ii++) {
			echo("Adding beer no "+Integer.toString(ii)+" to the bulk request.");
			Beer beer = makeABeer();
			String jsonString = mapper.writeValueAsString(beer);
//			IndexRequest irq =  new IndexRequest(indexName, typeName, "beer_"+Integer.toString(ii));
			
			IndexRequest irq = client.prepareIndex(indexName, typeName).setSource(jsonString).request();
			if (irq == null) {
				throw new Exception("The IndexRequest instance was null!!!");
			}
			brb.add(irq);
			echo("DONE Adding beer no "+ii+" to the bulk request.");
		}
		BulkResponse br = brb.execute().actionGet();
		if (br.hasFailures()) {
			throw new Exception("The bulk indexing of 10 beers had some failures!!!");
		}
		
		for (BulkItemResponse bulkItemResponse : br) {
			IndexResponse ir = bulkItemResponse.getResponse();
			String beerID = ir.getId();
			GetResponse gr = client.prepareGet(indexName, typeName, beerID).execute()
					.actionGet();
			Beer retrievedBeer = mapper.readValue(gr.getSourceAsBytes(), Beer.class);
			newBeers.put(beerID, retrievedBeer);
		}		
	}
	
	
	public static void retrieveAndCheckAllNewBeers() throws Exception {
		
		Iterator newBeersIt = newBeers.entrySet().iterator();
		while (newBeersIt.hasNext()) {
			Map.Entry pair = (Map.Entry)newBeersIt.next();
		    String beerId = (String)pair.getKey();
		    Beer beer = (Beer)pair.getValue();
			echo("Retrieving beer object with ID="+beerId+"...");
			GetResponse gr = client.prepareGet(indexName, typeName, beerId).execute()
					.actionGet();
			Beer retrievedBeer = mapper.readValue(gr.getSourceAsBytes(), Beer.class);
			if (!retrievedBeer.equals(beer)) {
				throw new Exception("Beer that was retrieved using ID="+beerId+" was not identical to the one that was originally indexed!!!");
			}
			echo("DONE Retrieving beer object with ID="+beerId+"...");
		}			
	}
	
	public static void deleteAllButOneOfTheNewBeers() {
		Iterator newBeersIt = newBeers.entrySet().iterator();
		boolean isFirst = true;
		while (newBeersIt.hasNext()) {
			Map.Entry pair = (Map.Entry)newBeersIt.next();
			if (!isFirst) {
			    String beerId = (String)pair.getKey();
				echo("Deleting beer object with ID="+beerId+"...");
				DeleteResponse dr = client.prepareDelete(indexName, typeName, beerId).execute().actionGet();
				echo("DONE Deleting beer object with ID="+beerId+"...");
			}
			isFirst = false;
		}					
	}
	
	public static Beer makeABeer() throws JsonGenerationException, JsonMappingException, IOException {
		String beerName = createUniqueBeerName();
		echo("Creating a beer object with name="+beerName+"...");
		Beer beer = new Beer(beerName, Colour.PALE, 0.33, 3);
		echo("DONE Creating a beer object with name="+beerName+"...");
		
		return beer;
	}
	
	public static float elapsedSecs() {
		float elapsed = (System.currentTimeMillis() - startTimeMSecs)/1000;
		return elapsed;
	}
	
	public static void echo(String mess) {
		mess = mess + " (Elapsed so far: "+elapsedSecs()+" seconds)";
		System.out.println(mess);
	}
	
	public static String createUniqueBeerName() {
		long now = System.currentTimeMillis();
		String beerName = "Heineken"+Long.toString(now);
		return beerName;
	}	
}