gistfile1.java

import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.node.Node;
import org.elasticsearch.node.NodeBuilder;


public class Test {

    static final String updateScript =
            "int i=0; " +
                    "found=false; " +
                    "while (!found && i < ctx._source.names.size()) { " +
                    "if (ctx._source.names.get(i).value == name) { " +
                    "found = true; " +
                    "} " +
                    "i++; " +
                    "} " +
                    "if (found) { " +
                    "ctx._source.names.get(i-1).count += 1; " +
                    "} " +
                    "else { " +
                    "ctx._source.names.add({'value' : name, 'count' : 1 }); " +
                    "} ";

    static final String indexName = "sample-index";

    static final String indexType = "sample";

    private static class ElasticUpdate implements Runnable {
        public void run() {
            for (int i = 0; i < 100000; i++) {
                String name = "name" + (i % 100);
                String key = "key" + (i % 11);
                //System.out.println("Thread "+Thread.currentThread().getName()+" updating "+key+" with name "+name);
                GetResponse getResponse = client.prepareGet(indexName, indexType, key).execute().actionGet();
                if (!getResponse.exists()) {
                    client.prepareIndex(indexName, indexType, key).setSource("{ \"names\" : [] }")
                            .execute()
                            .actionGet();
                }
                client.prepareUpdate(indexName, indexType, key)
                        .setRetryOnConflict(20)
                        .setScript(updateScript)
                        .addScriptParam("name", name)
                        .execute()
                        .actionGet();
            }
        }

        Client client;

        ElasticUpdate(Client client) {
            this.client = client;
        }
    }

    public static void main(String args[]) throws InterruptedException {
        Node node = NodeBuilder.nodeBuilder().local(true)
                .settings(ImmutableSettings.settingsBuilder()
                        .put("gateway.type", "none")
                )
                .node();
        node.client().admin().indices().prepareDelete().execute().actionGet();
        node.client().admin().indices().prepareCreate(indexName)
                .setSettings(ImmutableSettings.settingsBuilder().put("index.number_of_shards", 1).put("index.number_of_replicas", 0))
                .addMapping(indexType, "{\n" +
                        "    \"sample\" : {\n" +
                        "       \"properties\" : {\n" +
                        "          \"names\": {\n" +
                        "            \"properties\" : {\n" +
                        "              \"value\": {  \"type\" : \"string\", \"index\" : \"no\" },\n" +
                        "              \"count\": { \"type\" : \"integer\", \"index\" : \"no\" }\n" +
                        "            }\n" +
                        "          }\n" +
                        "       }\n" +
                        "    }\n" +
                        "  }")
                .execute().actionGet();
        node.client().admin().cluster().prepareHealth().setWaitForGreenStatus().execute().actionGet();
        Client client = node.client();


        int nthreads = 4;
        if (args.length > 0) {
            try {
                nthreads = Integer.parseInt(args[0]);
            } catch (NumberFormatException e) {
                System.err.println("Argument must be an integer.");
                System.exit(1);
            }
        }
        System.out.println("Starting " + nthreads + " threads");
        for (int i = 0; i < nthreads; i++) {
            Thread t = new Thread(new ElasticUpdate(client));
            t.start();
        }
    }
}