ElasticSearch javaAPI demo示例

缺乏、安全感 2022-06-08 14:19 266阅读 0赞

### 欢迎关注“**程序杂货铺**”公众号，里面有精彩内容，欢迎大家收看^\_^ ###

# 一.数据实体： #

![Center][]

ID：自增主键

ID：ID值

VALUE\_DAY：该机器当天的值

GROUPNAME：组名

TYPE\_MACHINE：机器类型

DATE：日期

TYPE：类型

# 二.Java代码： #

## 1.创建客户端： ##

package elasticsearch.util;
    
    import java.net.InetAddress;
    
    import org.elasticsearch.client.Client;
    import org.elasticsearch.client.transport.TransportClient;
    import org.elasticsearch.common.transport.InetSocketTransportAddress;
    
    
    public class ClientUtil {
    	private Client client;
    	public ClientUtil() {
    		try {
    			init();
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    	}
    	
    	/**
    	 * 初始化客户端
    	 * */
    	private void init() throws Exception {
    		byte[] bs = new byte[] { (byte) 192, (byte) 168, (byte)52, (byte)130 }; 
    		this.client = TransportClient.builder().build()  
                    .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByAddress(bs), 9300));
    	}
    
    	public Client getClient() {
    		return client;
    	}
    
    }

## 2.创建索引，主要创建mapper ##

package elasticsearch.service;
    
    import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
    import org.elasticsearch.client.Client;
    import org.elasticsearch.client.IndicesAdminClient;
    import org.elasticsearch.common.settings.Settings;
    import org.elasticsearch.common.xcontent.XContentBuilder;
    import org.elasticsearch.common.xcontent.XContentFactory;
    
    public class CreateIndex {
    	
    	/**
    	 * 创建空索引  默认setting 无mapping
    	 * @param client
    	 * @param index
    	 * @return
    	 */
    	public boolean createSimpleIndex(Client client, String index){
    	    IndicesAdminClient indicesAdminClient = client.admin().indices();
    	    CreateIndexResponse response = indicesAdminClient.prepareCreate(index).get();
    	    return response.isAcknowledged();
    	}
    	
    	/**
    	 * 创建索引 指定setting,创建mapper
    	 * @param client
    	 * @param index
    	 * @return
    	 */
    	public boolean createIndex(Client client, String index){
    	    // settings
    	    Settings settings = Settings.builder()
    	    		.put("index.number_of_shards", 1)
    	    		.put("index.number_of_replicas", 0)
    	    		//.put("cluster.name", "poc")
    	    		//.put("node.name", "node1")
    	    		//.put("client.transport.ignore_cluster_name", true)
    	    		//.put("node.client", true)
    	    		//.put("client.transport.sniff", true)
    	    		.build();
    	    // mapping
    	    XContentBuilder mappingBuilder;
    	    try {
    	        mappingBuilder = XContentFactory.jsonBuilder()
    	                .startObject()
    	                    .startObject(index)
    	                        .startObject("properties")
    	                            .startObject("ID").field("type", "string").field("store", "yes").endObject()
    	                            .startObject("IP").field("type", "string").field("store", "yes").endObject()
    	                            .startObject("VALUE_DAY").field("type", "string").field("store", "yes").endObject()
    	                            .startObject("GROUPNAME").field("type", "string").field("store", "yes").endObject()
    	                            .startObject("TYPE_MACHINE").field("type", "string").field("store", "yes").endObject()
    	                            .startObject("DATE").field("type", "string").field("store", "yes").endObject()
    	                            .startObject("TYPE").field("type", "string").field("store", "yes").endObject()
    	                        .endObject()
    	                    .endObject()
    	                .endObject();
    	    } catch (Exception e) {
    	        System.out.println("--------- createIndex 创建 mapping 失败：" + e);
    	        return false;
    	    }
    	    IndicesAdminClient indicesAdminClient = client.admin().indices();
    	    CreateIndexResponse response = indicesAdminClient.prepareCreate(index)
    	            .setSettings(settings)
    	            .addMapping(index, mappingBuilder)
    	            .get();
    	    return response.isAcknowledged();
    	}
    }

## 3.将元数据写入索引文件 ##

### (1)获取元数据 ###

package elasticsearch.util;
    
    import java.io.BufferedReader;
    import java.io.FileReader;
    import java.util.ArrayList;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    
    import net.sf.json.JSONObject;
    
    public class DataFactory {
    	public static List<String> dataCreater(String path) throws Exception {
    		
    		FileReader fr = new FileReader(path);
    		BufferedReader br = new BufferedReader(fr);
    		String line=null;
    		List<String> dataList = new ArrayList<String>();
    		
    		while((line = br.readLine()) != null){
    			String[] data = line.split("\t");
    			Map<String, String> map = new HashMap<String, String>();
    			map.put("ID", data[0]);
    			map.put("IP", data[1]);
    			map.put("VALUE_DAY", data[2]);
    			map.put("GROUPNAME", data[3]);
    			map.put("TYPE_MACHINE", data[4]);
    			map.put("DATE", data[5]);
    			map.put("TYPE", data[6]);
    			JSONObject jsonObject = JSONObject.fromObject(map);
    			String str = jsonObject.toString();
    			dataList.add(str);
    		}
    		
    		br.close();
    		fr.close();
    		
    		return dataList;
    	}
    }

### (2)数据写入索引 ###

package elasticsearch.service;
    
    import java.util.List;
    
    import org.elasticsearch.action.index.IndexRequestBuilder;
    import org.elasticsearch.action.index.IndexResponse;
    import org.elasticsearch.client.Client;
    
    import elasticsearch.util.DataFactory;
    
    public class WriteToIndex {
    	
    	/**
    	 * 数据写入索引
    	 * @param client
    	 * @param index
    	 * @return
    	 * @throws Exception 
    	 */
    	public void write(Client client, String index, String tablename, String filePath) throws Exception{
    		
    		//创建索引库 需要注意的是.setRefresh(true)这里一定要设置,否则第一次建立索引查找不到数据
            IndexRequestBuilder requestBuilder = client.prepareIndex(index, tablename).setRefresh(true);
    		List<String> dataList = DataFactory.dataCreater(filePath);
    	    for(String data : dataList){
    	    	//System.out.println(data);
    	    	requestBuilder.setSource(data).execute().actionGet();
    	    }
    	}
    }

## 4.查询 ##

package elasticsearch.service;
    
    import java.util.ArrayList;
    import java.util.List;
    
    import org.elasticsearch.action.search.SearchResponse;
    import org.elasticsearch.client.Client;
    import org.elasticsearch.common.unit.TimeValue;
    import org.elasticsearch.index.query.QueryBuilder;
    import org.elasticsearch.index.query.QueryBuilders;
    import org.elasticsearch.search.SearchHit;
    import org.elasticsearch.search.SearchHits;
    
    public class SearchIndex {
    	
    	/**
    	 * 查询全部内容，实现分页查询
    	 * */
    	public List<String> searchAll(Client client, String index, String table) {
    		
    		SearchResponse response = client.prepareSearch().setIndices(index).setTypes(table).get();
    		int n = (int) response.getHits().totalHits();
    		int size = 100;
    		List<String> retList = new ArrayList<String>();
    		
    		for(int i = 0;i < n;){
    			response = client.prepareSearch().setIndices(index).setTypes(table).setScroll(TimeValue.timeValueMinutes(2)).setFrom(i).setSize(size).get();
    			i += size;
    			SearchHits searchHits = response.getHits();
    			for (SearchHit hit : searchHits) {
    		        retList.add(hit.getSourceAsString());
    		    }
    		}
    		return retList;
    	}
    	
    	/**
    	 * 使用过滤器查询，实现分页查询
    	 * */
    	public List<String> queryByFilter(Client client, String index, String table) {
    		
    		// 查询groupname为"压力测试"的数据
    	    QueryBuilder queryBuilder = QueryBuilders.matchQuery("GROUPNAME", "压力测试");
    	    
    		SearchResponse response = client.prepareSearch().setIndices(index).setTypes(table).setQuery(queryBuilder).get();
    		int n = (int) response.getHits().totalHits();
    		System.out.println(n);
    		int size = 100;
    		List<String> retList = new ArrayList<String>();
    		
    		response = client.prepareSearch().setIndices(index).setTypes(table).setScroll(TimeValue.timeValueMinutes(5)).setSize(n).setQuery(queryBuilder).get();
    		SearchHits searchHits = response.getHits();
    		for (SearchHit hit : searchHits) {
    	        retList.add(hit.getSourceAsString());
    	    }
    		
    		return retList;
    	}
    
    }

## 5.测试方法 ##

package elasticsearch.test;
    
    import java.util.List;
    
    import org.elasticsearch.client.Client;
    
    import elasticsearch.service.CreateIndex;
    import elasticsearch.service.SearchIndex;
    import elasticsearch.service.WriteToIndex;
    import elasticsearch.util.ClientUtil;
    
    public class test {
    	
    	/**
    	 * 创建索引，主要创建mapper
    	 * */
    	public static void createIndex(ClientUtil clientUtil) {
    		CreateIndex test = new CreateIndex();
    		test.createIndex(clientUtil.getClient(), "poc");
    	}
    	
    	/**
    	 * 将元数据写入索引文件
    	 * */
    	public static void insertDataToIndex(ClientUtil clientUtil) {
    		WriteToIndex test = new WriteToIndex();
    		try {
    			test.write(clientUtil.getClient(), "poc", "zabbixmetadata", "data/poc");
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    	}
    	
    	/**
    	 * 查询
    	 * */
    	public static void searchIndex(ClientUtil clientUtil) {
    		SearchIndex search = new SearchIndex();
    		//List<String> ret = search.searchAll(clientUtil.getClient(), "poc", "zabbixmetadata");
    		List<String> ret = search.queryByFilter(clientUtil.getClient(), "poc", "zabbixmetadata");
    		System.out.println(ret.size());
    		System.out.println(ret);
    	}
    
    	public static void main(String[] args) throws Exception {
    		ClientUtil clientUtil = new ClientUtil();
    		//createIndex(clientUtil);        // 1.创建索引文件
    		//insertDataToIndex(clientUtil);  // 2.将元数据写入索引
    		//searchIndex(clientUtil);        // 3.查询
    	}
    
    }

## 6.备注 ##

详细源代码请见：

http://download.csdn.net/download/u013473512/9998225

https://github.com/Emmitte/ElasticSearch

欢迎关注“ **程序杂货铺**”公众号，里面有精彩内容，欢迎大家收看^\_^

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTM0NzM1MTI_size_16_color_FFFFFF_t_70][]

[Center]: /images/20220608/f59f9199cea64513babda216df3bd106.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTM0NzM1MTI_size_16_color_FFFFFF_t_70]: /images/20220608/5b51d3c3b32942b5915d0d72d1a4feb4.png