@@ -0,0 +1,2267 @@
+package reactiontest.elastic;%0A%0Aimport java.net.InetAddress;%0Aimport java.net.UnknownHostException;%0A%0Aimport org.elasticsearch.action.search.SearchResponse;%0Aimport org.elasticsearch.client.Client;%0Aimport org.elasticsearch.client.transport.TransportClient;%0Aimport org.elasticsearch.common.settings.Settings;%0Aimport org.elasticsearch.common.transport.InetSocketTransportAddress;%0Aimport org.elasticsearch.index.query.QueryBuilders;%0Aimport org.elasticsearch.search.SearchHit;%0A%0A%0Apublic class ElasticSearch %7B%0A%09%0A%09private Client client = null;%0A%09%0A%09public ElasticSearch() throws UnknownHostException %7B%0A%09%09Settings settings = Settings.settingsBuilder()%0A%09%09        .put(%22cluster.name%22, %22my-demo%22).build();%0A%09%09client = TransportClient.builder().settings(settings).build()%0A%09%09%09%09.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(%22localhost%22), 9300));%0A%09%7D%0A%0A%09public void search()%7B%0A%09%09//curl 'localhost:9200/viper-test/viper-log/_search?regina'%0A%09%09SearchResponse response = client.prepareSearch(%22viper-test%22)%0A%09%09        .setTypes(%22viper-log%22)%0A%09%09        .setQuery(QueryBuilders.matchQuery(%22second%22, %22regina%22))%0A%09%09        //.setSearchType(SearchType.DFS_QUERY_THEN_FETCH)%0A%09%09        //.setQuery(QueryBuilders.termQuery(%22regina%22, %22test%22))                 // Query%0A%09%09        //.setPostFilter(QueryBuilders.rangeQuery(%22age%22).from(12).to(18))     // Filter%0A%09%09        //.setFrom(0).setSize(60).setExplain(true)%0A%09%09        .execute()%0A%09%09        .actionGet();%0A%09%09System.out.println(response);%0A        SearchHit%5B%5D results = response.getHits().getHits();%0A        for(SearchHit hit : results)%7B%0A%0A            String sourceAsString = hit.getSourceAsString();%0A            if (sourceAsString != null) %7B%0A                //Gson gson = new GsonBuilder().setDateFormat(dateFormat)%0A                        //.create(); %0A            %09System.out.println(sourceAsString);%0A                //System.out.println( gson.fromJson(sourceAsString, Firewall.class));%0A            %7D%0A        %7D%0A%09%09//GetResponse response2 = client.prepareGet(%22viper-test%22, %22viper-log%22, %22regina%22).get();%0A%09%09//System.out.println(response2);%0A%09%7D%0A%09%0A%09public void closeConenction()%7B%0A%09%09client.close();%0A%09%7D%0A%09%0A%09public static void main(String%5B%5D args) throws UnknownHostException %7B%0A%09%09ElasticSearch e = new ElasticSearch();%0A%09%09e.search();%0A%09%09%0A%09%7D%0A%09%0A%7D%0A

add elasticsearch source to create flink DataSet