Elasticsearch创建Index--java实现

Posted 2022-12-13 ZK_小姜

将所要创建的index以json格式的方式写入配置文件，可以指定所要创建的index的字段类型，然后通过java api来创建index

具体实现：

Index对象类：

public class Index 

	private String index;                   //索引名
	private String type;					//type表名	
	private Integer number_of_shards;		//分片数
	private Integer number_of_replicas;		//备份数
	private String fieldJson;				//字段类型
	public String getIndex() 
		return index;
	
	public void setIndex(String index) 
		this.index = index;
	
	public String getType() 
		return type;
	
	public void setType(String type) 
		this.type = type;
	
	public String getFieldJson() 
		return fieldJson;
	
	public void setFieldJson(String fieldJson) 
		this.fieldJson = fieldJson;
	
	public Integer getNumber_of_shards() 
		return number_of_shards;
	
	public void setNumber_of_shards(Integer number_of_shards) 
		this.number_of_shards = number_of_shards;
	
	public Integer getNumber_of_replicas() 
		return number_of_replicas;
	
	public void setNumber_of_replicas(Integer number_of_replicas) 
		this.number_of_replicas = number_of_replicas;
	
	

elasticsearch.properties配置文件：

hosts:172.17.6.203:9300,172.17.6.202:9300
clusterName:log-test

index.json配置文件：

[
	"index":"index1",
	"type":"initType",
	"number_of_shards":3,
	"number_of_replicas":2,
	"fieldType":
		"initType":
			"properties":
			"name":
				"type":"String"
			,
			"date":
				"type":"date",
				"format": "yyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis"
			
		
		
	
,
	"index":"index2",
	"type":"initType",
	"number_of_shards":3,
	"number_of_replicas":2,
	"fieldType":
		"initType":
			"properties":
			"ip":
				"type":"ip"
			,
			"date":
				"type":"date",
				"format": "yyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis"
			
		
		
	
]

创建index 具体类：

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.IndicesAdminClient;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

public class Elasticsearch 

	private Client client;
	private IndicesAdminClient adminClient;
	/**
	 * 集群配置初始化方法
	 * @throws Exception
	 */
	private void init() throws Exception
		Properties properties = readElasticsearchConfig();
		String clusterName = properties.getProperty("clusterName");
		String[] inetAddresses = properties.getProperty("hosts").split(",");
		Settings settings = Settings.settingsBuilder().put("cluster.name", clusterName)
				.build();
		client = TransportClient.builder().settings(settings)
				.build();
		for (int i = 0; i < inetAddresses.length; i++) 
			String[] tmpArray = inetAddresses[i].split(":");
			String ip = tmpArray[0];
			int port = Integer.valueOf(tmpArray[1]);
			client = ((TransportClient)client).addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(ip), port));
		
	
	/**
	 * 构造方法
	 */
	public Elasticsearch() 
		try 
			init();
		 catch (Exception e) 
			System.out.println("init() exception!");
			e.printStackTrace();
		
		adminClient = client.admin().indices();
	
	/**
	 * 判断集群中index是否存在
	 * @param index
	 * @return 存在（true）、不存在（false）
	 */
	public boolean indexExists(String index)
		IndicesExistsRequest request = new IndicesExistsRequest(index);
		IndicesExistsResponse response = adminClient.exists(request).actionGet();
		if (response.isExists()) 
			return true;
		
		return false;
	
	/**
	 * 读取es配置信息
	 * @return
	 */
	private Properties readElasticsearchConfig() 
		Properties properties = new Properties();
		try 
			InputStream is = this.getClass().getClassLoader().getResourceAsStream("elasticsearch.properties");
			properties.load(new InputStreamReader(is,"UTF-8"));
		 catch (IOException e) 
			System.out.println("readEsConfig exception!");
			e.printStackTrace();
		
		return properties;
	
	/**
	 * 读取json配置文件
	 * @return JSONArray
	 * @throws IOException
	 */
	public JSONArray readJosnFile() throws IOException
		InputStream is = this.getClass().getClassLoader().getResourceAsStream("index.json");
		BufferedReader br = new BufferedReader(new InputStreamReader(is, "UTF-8"));
		StringBuffer sb = new StringBuffer();
		String tmp = null;
		while((tmp=br.readLine()) != null)
			sb.append(tmp);
		
		JSONArray result = JSONArray.parseArray(sb.toString());
		return result;
	
	/**
	 * 获取要创建的index列表
	 * @param client
	 * @return List<Index>
	 */
	public List<Index> getIndexList()
		
		List<Index> result = new ArrayList<Index>();
		JSONArray jsonArray = null;
		try 
			jsonArray = readJosnFile();
		 catch (IOException e) 
			System.out.println("readJsonFile exception!");
			e.printStackTrace();
		
		if (jsonArray == null || jsonArray.size() == 0) 
			return null;
		
		for (int i = 0; i < jsonArray.size(); i++) 
			JSONObject jsonObject = jsonArray.getJSONObject(i);
			Index indexObject = new Index();
			String index = jsonObject.getString("index");
			String type = jsonObject.getString("type");
			Integer number_of_shards = jsonObject.getInteger("number_of_shards");
			Integer number_of_replicas = jsonObject.getInteger("number_of_replicas");
			String fieldRType = jsonObject.get("fieldType").toString();
			indexObject.setIndex(index);
			indexObject.setType(type);
			indexObject.setFieldJson(fieldRType);
			indexObject.setNumber_of_shards(number_of_shards);
			indexObject.setNumber_of_replicas(number_of_replicas);
			result.add(indexObject);
		
		return result;
	
	/**
	 * 创建Index
	 * @param client
	 */
	public void CreateIndex()
		int i = 0;
		List<Index> list = getIndexList();
		IndicesAdminClient adminClient = client.admin().indices();
		for(Index index : list)
			if (indexExists(index.getIndex())) 
				continue;
			
			adminClient.prepareCreate(index.getIndex())
			.setSettings(Settings.builder().put("index.number_of_shards", index.getNumber_of_shards()).put("index.number_of_replicas", index.getNumber_of_replicas()))
			.addMapping(index.getType(), index.getFieldJson())
			.get();
			i++;
		
		System.out.println("index creation success! create "+i+" index");
		
	
	public static void main(String[] args) 
		Elasticsearch es = new Elasticsearch();
		es.CreateIndex();