Elasticsearch for Java API Simple Example

ElasticsearchJavaExample

/**
 * Elasticsearch Java API Example
 */

import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.admin.indices.mapping.delete.DeleteMappingResponse;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse;
import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsResponse;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.count.CountResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHitField;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;

import static org.elasticsearch.index.query.QueryBuilders.*;

/**
 * Created by SteveShin on 2015-05-08.
 */
public class EsSample {

    Logger logger = LoggerFactory.getLogger(EsSample.class);

    static final ForkJoinPool forkJoinPool = new ForkJoinPool();
    static final int bytes = 1000;
    static String indexName = "size_sample";
    static String indexType = "sample";
    static int amount = 1000000;

    static final Map<String, Object> beforeSet = new HashMap<String, Object>(){
        {
            put("index.refresh_interval",-1);
            put("index.number_of_replicas",0);
        }
    };
    static final Map<String, Object> afterSet = new HashMap<String, Object>(){
        {
            put("index.refresh_interval","10s");
            put("index.number_of_replicas",0);
        }
    };
    //create Index(schema)
    public static void createIndex(Client client, String indexName) {
        System.out.println(indexName);
        IndicesExistsResponse res = client.admin().indices().prepareExists(indexName).execute().actionGet();
        if (!res.isExists()) {
            CreateIndexRequestBuilder createIndexRequestBuilder = client.admin().indices().prepareCreate(indexName);
            createIndexRequestBuilder.setSettings(beforeSet);
            CreateIndexResponse response = createIndexRequestBuilder.execute().actionGet();
            if(response.isAcknowledged()) {
                String mappingJson = "{" +
                        "\"_all\":{\"enabled\":false}" +
                        ",\"_source\":{\"enabled\":false}" +
                        ",\"properties\":{" +
                            "\"body\":{\"type\":\"string\",\"store\":true}" +
                            ",\"head\":{\"type\":\"object\",\"store\":false}" +
                        "}}";
                PutMappingResponse mappingResponse = client.admin().indices()
                        .preparePutMapping(indexName)
                        .setType(indexType)
                        .setSource(mappingJson)
                        .execute().actionGet();
                System.out.println("mapping... " + mappingResponse.isAcknowledged());
            }
            System.out.println("CREATE index: " + response.isAcknowledged());
        }
    }

    //delete document
    public static void deleteDocument(Client client, String indexName, String typeName, String documentId) {
        DeleteResponse response = client.prepareDelete(indexName, typeName, documentId)
                .execute()//.setOperationThreades(false)//run on this thread
                .actionGet();
    }

    //delete type
    public static void deleteType(Client client, String indexName, String typeName) {
        DeleteMappingResponse response = client.admin().indices().prepareDeleteMapping(indexName).setType(typeName).execute().actionGet();

    }

    public static List<String> pumpup(List<String> logs, int loop) {
        List<String> result = new ArrayList<String>();
        System.out.println("Loop:" + loop + ", total size: " + logs.size() * loop);
        for (int i = 0; i < loop; i++) {
            result.addAll(logs);
        }
        return result;
    }

    public static void putTest(Client client) throws IOException, InterruptedException, ExecutionException {
        //prepare Bulk Processor
        BulkProcessor bulkProcessor = BulkProcessor.builder(client,
                new BulkProcessor.Listener() {
                    @Override
                    public void beforeBulk(long executionId, BulkRequest request) {
                        //System.out.println("Bulk process ready... : { id:" + executionId + ", size:" + request.requests().size() + "}\t");
                    }

                    @Override
                    public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                        System.out.println(" done... { id:" + executionId + ", time:" + response.getTookInMillis()
                                + ", size:" + response.getItems().length + "}");
                    }

                    @Override
                    public void afterBulk(long executionId, BulkRequest request, Throwable fail) {
                        fail.printStackTrace();
                        System.out.println(" fail...  { id:" + executionId + ", error:" + fail.getMessage() + "}");
                    }
                })
                .setBulkActions(10000)
                .setBulkSize(new ByteSizeValue(200, ByteSizeUnit.MB))
                .setFlushInterval(TimeValue.timeValueSeconds(10))
                .setConcurrentRequests(10)
                .build();
    }

    public static void indexSetting(Client client, Map<String, Object> settings){
        UpdateSettingsResponse response = client.admin().indices().prepareUpdateSettings(indexName).setSettings(settings).execute().actionGet();
        System.out.println("update set:" + response.isAcknowledged());
    }

    public static void put(Client client) throws IOException, InterruptedException, ExecutionException {
/*    //get indice(schema)
    GetMappingsResponse res = client.admin().indices().getMappings(new GetMappingsRequest().indices(indexName)).actionGet();
    ImmutableOpenMap<String, MappingMetaData> mapping = res.mappings().get(indexName);
    for (ObjectObjectCursor<String, MappingMetaData> c : mapping) {
        System.out.println("index find: " + c.key + " = " + c.value.source());
    }*/

//        indexSetting(client, beforeSet);
        //indexSetting(client, beforeSet);
        //CreateIndexResponse createIndexResponse = client.admin().indices().create(new CreateIndexRequestBuilder. .createIndexRequest()).actionGet();

        //prepare Bulk Processor
        BulkProcessor bulkProcessor = BulkProcessor.builder(client,
                new BulkProcessor.Listener() {
                    @Override
                    public void beforeBulk(long executionId, BulkRequest request) {
                        //System.out.println("Bulk process ready... : { id:" + executionId + ", size:" + request.requests().size() + "}\t");
                    }

                    @Override
                    public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                        System.out.println(" done... { id:" + executionId + ", time:" + response.getTookInMillis()
                                + ", size:" + response.getItems().length + "}");
                    }

                    @Override
                    public void afterBulk(long executionId, BulkRequest request, Throwable fail) {
                        fail.printStackTrace();
                        System.out.println(" fail...  { id:" + executionId + ", error:" + fail.getMessage() + "}");
                    }
                })
                .setBulkActions(10000)
                .setBulkSize(new ByteSizeValue(150, ByteSizeUnit.MB))
                //.setFlushInterval(TimeValue.timeValueSeconds(120))
                .setConcurrentRequests(3)
                .build();

        //get sample logs
        List<String> message = pumpup(TtDummyRvMsgGenerate.read(), 1);
        TtRvMsgParser parser = new TtRvMsgParser();

        List<Map<String, Object>> listMapList = new ArrayList<>();
        for (String msg : message) {
            parser.init(msg);//.println();
            Map<String, Object> listMap = parser.result();
            listMapList.add(listMap);
        }

        int loop = 1;
        long start = new Date().getTime();
        while (loop <= amount) {
            //use not parsed
//            for(final String msg : message){
//                @SuppressWarnings("unchecked")
//                Map<String, String> fullMsg = new HashMap<String, String>(){{
//                    put("body", TtDummyRvMsgGenerate.increaseByte(msg, 1000));
//                }};
//                bulkProcessor.add(new IndexRequest(indexName, indexType).source(fullMsg));
//                loop++;
//            }
            //use parsed
            for (Map<String, Object> listMap : listMapList) {
                //String a = (String)listMap.get("body");
                //System.out.println(a);
                //listMap.put("body", ((String) listMap.get("body")).concat(TtDummyRvMsgGenerate.getSampleWords(15)));
                //listMap.put("body", ((String) listMap.get("body")).concat(TtDummyRvMsgGenerate.randomKey(15)));
                //listMap.put("body", TtDummyRvMsgGenerate.increaseByte((String) listMap.get("body"), 500));

                //listMap.put("loop", loop);
                bulkProcessor.add(new IndexRequest(indexName, indexType, loop+"").source(listMap));
                loop++;
            }
            //System.exit(0);
        }
        long end = new Date().getTime();
        System.out.println("-------------------- ELAPSED:" + (end - start) + " millis");
        indexSetting(client, afterSet);

        //Bulk Process close
        bulkProcessor.awaitClose(10, TimeUnit.MINUTES);
    }

    public static void getDocument(Client client, String indexName, String typeName, String documentId) {
        GetResponse getResponse = client.prepareGet(indexName, typeName, documentId).execute().actionGet();
        Map<String, Object> source = getResponse.getSource();
        System.out.println("------------------------------");
        System.out.println("Index: " + getResponse.getIndex());
        System.out.println("Type: " + getResponse.getType());
        System.out.println("Id: " + getResponse.getId());
        System.out.println("Version: " + getResponse.getVersion());
        System.out.println(source);
        System.out.println("------------------------------");
    }

    public static void fullTextSearchCount(Client client, String IndexName, String typeName, String value, String... field) {
        QueryBuilder queryBuilder =
                //moreLikeThisQuery(field).likeText(value).minTermFreq(1).maxQueryTerms(1000);
                //wildcardQuery(field[0], value);
                queryStringQuery(field[0] + ":" + value);
        CountResponse response = client.prepareCount(IndexName).setTypes(typeName)
                .setQuery(queryBuilder)
                .execute().actionGet();
        System.out.println("\tCurrent More Like This count: " + response.getCount() + "");
    }

    public static void fullTextSearchDocument(Client client, String indexName, String typeName, String value, String... field) {
        QueryBuilder fieldQueryBuilder =
                //termQuery(field[0], value);
                matchQuery(field[0], value);
                //new WildcardQueryBuilder(field[0], value);
                //moreLikeThisQuery(field).likeText(value).minTermFreq(1).maxQueryTerms(25);
                //wildcardQuery(field[0], value);
                //queryStringQuery(field[0] + ":" + value);
                //multiMatchQuery(value, field);
        //SortBuilder sortBuilder = SortBuilders.fieldSort(field[0]);
        SearchResponse searchResponse = client.prepareSearch(indexName).setTypes(typeName).addFields(field[0])
                .setSearchType(SearchType.QUERY_THEN_FETCH)//.addSort(sortBuilder)
                .setQuery(fieldQueryBuilder)//.addSort(SortBuilders.fieldSort("loop").order(SortOrder.DESC))
                .setFrom(0).setSize(1000).setExplain(false).execute().actionGet();

        SearchHit[] result = searchResponse.getHits().getHits();

        System.out.println(searchResponse.getHits().getTotalHits() + " total count");
        //get result & print
        System.out.println("\tCurrent More Like This results: " + result.length + "");
        for (SearchHit hit : result) {
            Set<String> keys = hit.fields().keySet();
            Iterator<String> keyItr = keys.iterator();
            while(keyItr.hasNext()){
                SearchHitField hitField = hit.field(keyItr.next());
                System.out.println(hitField.getValue().toString());
            }
//            Map<String, Object> r = hit.getSource();
//            System.out.println(r);
        }
    }

    public static boolean isOpen(Client client, String indexName){
        ClusterState clusterState = client.admin().cluster().prepareState().setIndices(indexName).get().getState();
        IndexMetaData metaData = clusterState.getMetaData().index(indexName);
        final String open = "OPEN";
        return metaData.getState() != null && metaData.getState().toString().equals(open);
    }

    public static void searchDocument(Client client, String indexName, String typeName, String value, String... field) {

        //query using query builder
        QueryBuilder matchQuery = matchQuery(field[0], value);
        //System.out.println(matchQuery.toString()); //query json println
        QueryBuilder multiMatchQuery = multiMatchQuery(value, field);
        QueryBuilder stringQuery = queryStringQuery(field[0] + ":" + value);
        //search query execution
        SearchResponse searchResponse = client.prepareSearch(indexName)
                .setTypes(typeName)
                .setSearchType(SearchType.QUERY_THEN_FETCH) //search type. hold
                .setQuery(stringQuery)
                .setFrom(0).setSize(1000) //limit
                .setExplain(false) //query explain
                .execute().actionGet(); //execute
        SearchHit[] result = searchResponse.getHits().getHits();

        //get result & print
        System.out.println("Current Match results: " + result.length + "--------------------");
        for (SearchHit hit : result) {
            Map<String, Object> r = hit.getSource();
            //System.out.println(r);
        }
    }

    public static void count(Client client, String index, String type) {
        CountResponse response = client.prepareCount(index)
                .setQuery(termQuery("_type", type))
                .execute().actionGet();
        System.out.println("type:" + type + "`s total count: " + response.getCount());
    }


    public static void main(String[] args) throws InterruptedException, ExecutionException, IOException {
        //ES connection set
        Settings settings = ImmutableSettings.settingsBuilder()
                .put("action.auto_create_index", true)
                .put("client.transport.sniff", true)
                .put("client.transport.ping_timeout", "60s")
                .put("network.tcp.block", true)
                //.put("node.name", "ES3")
                .put("cluster.name", "elasticsearch").build();

        //ES client
        //default native transport : 9300
        //default web/rest port : 9200
        Client client = new TransportClient(settings)
                .addTransportAddresses(
//                        new InetSocketTransportAddress("192.168.10.229", 9300)
                        new InetSocketTransportAddress("192.168.10.210", 9300)
                        , new InetSocketTransportAddress("192.168.10.251", 9300)
                        , new InetSocketTransportAddress("192.168.10.252", 9300)
                );
//        try {

//        ClusterStatsResponse res = client.admin().cluster().prepareClusterStats().execute().actionGet();
//
//        PutRepositoryResponse putRepositoryResponse = client.admin().cluster().preparePutRepository("hdfs_repo1")
//                .setType("hdfs")
//                .setSettings(
//                        ImmutableSettings.builder()
//                                .put("uri", "hdfs://192.168.10.251:2181/")
//                                .put("path", "es/snapshot")
//                                .put("concurrent_streams", 4)
//                                .put("compress", true)
//                                .put("chunk_size", "40mb")).get();
//        System.out.println("repository put response:" + putRepositoryResponse.isAcknowledged());
//
//        CreateSnapshotResponse createSnapshotResponse = client.admin().cluster().prepareCreateSnapshot("hdfs_repo1", "map1_2million")
//                .setIncludeGlobalState(false)
//                .setIndices("map1")
//                .setRepository("hdfs_repo1")
//                .setSettings(ImmutableSettings.builder().put("compress", true))
//                .setWaitForCompletion(true)
//                .get();
//        System.out.println("success:" + createSnapshotResponse.getSnapshotInfo().successfulShards());
//
//        GetSnapshotsResponse getSnapshotsResponse = client.admin().cluster().prepareGetSnapshots("hdfs_repo1").get();
//        ImmutableList<SnapshotInfo> snapshotInfos = getSnapshotsResponse.getSnapshots();
//        for(SnapshotInfo info : snapshotInfos){
//            System.out.println(info.name());
//        }
//
//        RestoreSnapshotResponse restoreSnapshotResponse = client.admin().cluster().prepareRestoreSnapshot("hdfs_repo1", "map1_2million")
//                .setIndexSettings(
//                        ImmutableSettings.builder().put("index.number_of_replicas", 0))
//                .setIndices("map1")
//                .setWaitForCompletion(true)
//                .setRenamePattern("map[.*]")
//                .setRenameReplacement("restore_map$1")
//                .setIgnoreIndexSettings("index.refresh_interval", "")
//                .get();
//
//
//        System.out.println(res.getClusterNameAsString());

            indexName = "disable_source";
        indexType = "test";
            amount = 10000;

        fullTextSearchDocument(client, indexName, indexType, "tmtm", "body");
        System.exit(0);
            //indexType = "test";
            createIndex(client, indexName);
            //Thread.sleep(1000);
            put(client);
            //System.out.println("loop : " + i);
        String repoName = "hdfs-repo";

        System.out.println("PUT END");



//        long start_ = new Date().getTime();
//        count(client, indexName, indexType);
//        long end_ = new Date().getTime();
//        System.out.println("--------------------------\nEnd count : " + (end_ - start_) + " time elapsed");
//        //getDocument(client, indexName, indexType, "1000");
//        //searchDocument(client, indexName, indexType, "155431.679016", "TXN_TIME", "TXN_KEY");
        List<String> query = Arrays.asList(
                "*ASDF1186*"
                , "*QLEGASD1800*"
                , "*WEFLKJASLD118*"
                , "*JGKLJDLSF1006*"
                , "*SLD2186*"
                , "*DLFKJQWEF1686*"
                , "*LKJFASD2718*"
                , "*LKJGQLEGASD235*"
                , "*KJQWEF1361*"
                , "*QWEFLKJASLD2*"
                , "*KLJGKLJDLSF ASDLKJFASD1*"
        );
        String index = "test3";
        String type = "rv_10000";
        //deleteType(client, "test", "ex_1000");
//        deleteType(client, "test2", "ex_10000_3");
//        System.out.println("delete complete");
//        System.exit(0);
        System.out.println(type + " full text search start");
        System.out.println("--------------------------------------------------------------------------------------------------");
        for (int i = 0; i <= 9; i++) {
            System.out.println("LOOP" + i + "test start");
            for (String q : query) {
                long start = new Date().getTime();
                String mergeQuery = q;
                //fullTextSearchCount(client, index, type, mergeQuery, "body");
                fullTextSearchDocument(client, index, type, mergeQuery, "body");
                long end = new Date().getTime();
                System.out.println("\t--------------End search : " + "query:" + mergeQuery + ", " + (end - start) + " time elapsed");
            }
            System.out.println("end of test---------------------------------------------------------------------------------------");
//            Thread.sleep(5000);
        }

//        long start1 = new Date().getTime();
//        fullTextSearchCount(client, indexName, indexType, " tmtm", "body");
//        long end2 = new Date().getTime();
//        System.out.println("--------------------------\nEnd count : " + (end2 - start1) + " time elapsed");
//        } catch (IOException | InterruptedException | ExecutionException e) {
//            e.printStackTrace();
//        }
        //Connection close
        client.close();
    }
}