先用java創(chuàng)建esClient

package com.ws.es.config;
import com.ws.util.PropertiesUtil;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.message.BasicHeader;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.client.Node;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;

import java.util.ArrayList;
import java.util.List;

public class ElasticSearchConfig {

    private static final Logger LOGGER = LogManager.getLogger(ElasticSearchConfig.class);

    /**
     * elk集群地址
     */
    private static String esServerHost = PropertiesUtil.getProperty("es.ServerHost");

    private static Integer connectTimeout = Integer.parseInt(PropertiesUtil.getProperty("es.connectTimeout"));

    private static Integer socketTimeout = Integer.parseInt(PropertiesUtil.getProperty("es.socketTimeout"));

    private static Integer connectionRequestTimeout = Integer
            .parseInt(PropertiesUtil.getProperty("es.connectionRequestTimeout"));

    private static Integer maxConnPerRoute = Integer.parseInt(PropertiesUtil.getProperty("es.maxConnPerRoute"));

    private static Integer maxConnTotal = Integer.parseInt(PropertiesUtil.getProperty("es.maxConnTotal"));

    private static boolean isSecureMode = Boolean.parseBoolean(PropertiesUtil.getProperty("es.isSecureMode"));

    /**
     * Bean name default 函數(shù)名字
     *
     * @return
     */
    public static RestClient restClient() {
        LOGGER.info("Elasticsearch初始化開始。。。。。");
        RestClientBuilder restClientBuilder = getRestClientBuilder();
        RestClient restClient = restClientBuilder.build();
        setNodes(restClient);
        LOGGER.info("The Low Level Rest Client has been created.");
        return restClient;
    }

    private static RestClientBuilder getRestClientBuilder() {
        HttpHost[] hostArray = getHostArray();
        if (!isSecureMode) {
            System.setProperty("es.security.indication", "false");
        } else {
            setSecConfig();
        }

        RestClientBuilder builder = RestClient.builder(hostArray);
        Header[] defaultHeaders = new Header[] { new BasicHeader("Accept", "application/json"),
                new BasicHeader("Content-type", "application/json") };
        builder.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
            @Override
            public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
                return requestConfigBuilder.setConnectTimeout(connectTimeout).setSocketTimeout(socketTimeout)
                        .setConnectionRequestTimeout(connectionRequestTimeout);
            }
        }).setMaxConnPerRoute(maxConnPerRoute).setMaxConnTotal(maxConnTotal);
        builder.setDefaultHeaders(defaultHeaders);
        return builder;
    }

    private static void setNodes(RestClient restClient) {
        List<Node> nodes = new ArrayList();
        HttpHost[] var3 = getHostArray();
        int var4 = var3.length;

        for (int var5 = 0; var5 < var4; ++var5) {
            HttpHost httpHost = var3[var5];
            nodes.add(new Node(httpHost));
        }
        restClient.setNodes(nodes);
    }

    private static HttpHost[] getHostArray() {
        String schema;
        if (!isSecureMode) {
            schema = "http";
        } else {
            schema = "https";
        }
        List<HttpHost> hosts = new ArrayList();
        String[] hostArray1 = esServerHost.split(",");
        String[] var4 = hostArray1;
        int var5 = hostArray1.length;

        for (int var6 = 0; var6 < var5; ++var6) {
            String host = var4[var6];
            String[] ipPort = host.split(":");
            HttpHost hostNew = new HttpHost(ipPort[0], Integer.parseInt(ipPort[1]), schema);
            hosts.add(hostNew);
        }

        return (HttpHost[]) hosts.toArray(new HttpHost[0]);
    }

    private static void setSecConfig() {
        try {
            System.setProperty("elasticsearch.kerberos.jaas.appname", "EsClient");
            System.setProperty("es.security.indication", "true");
        } catch (Exception var2) {
        }

    }
}

創(chuàng)建es索引模板

 public static void cateteTemplateNonMotor(RestClient restClient) {
        Map<String, String> params = Collections.singletonMap("pretty", "true");
        JSONObject json = new JSONObject();
        JSONObject settings = new JSONObject();
        json.put("template", "索引的模板前綴加*號例123*");
        settings.put("number_of_shards", "" + 分片數(shù)例2 + "");
        settings.put("number_of_replicas", "" + 副本數(shù)例1 + "");
        settings.put("max_result_window", "100000000");
        json.put("settings", settings);
        JSONObject mappings = new JSONObject();
        JSONObject clgjxx = new JSONObject();
        JSONObject properties = new JSONObject();
        properties.put("1", JSONObject.parse("{\"type\":\"keyword\"}"));// 
        properties.put("2", JSONObject.parse("{\"type\":\"keyword\"}"));// 
        properties.put("3", JSONObject.parse("{\"type\":\"keyword\"}"));// 
        properties.put("4", JSONObject.parse("{\"type\":\"keyword\"}"));// 
        properties.put("5", JSONObject.parse("{\"type\":\"keyword\"}"));// 
        properties.put("6", JSONObject.parse("{\"type\":\"keyword\"}"));// 
        properties.put("7", JSONObject.parse("{\"type\":\"keyword\"}"));// 
        properties.put("8", JSONObject.parse("{\"type\":\"keyword\"}"));//
        properties.put("9", JSONObject.parse("{\"type\":\"keyword\"}"));// 
        properties.put("10", JSONObject.parse("{\"type\":\"date\",\"format\":\"yyyyMMddHHmmss\"}"));// 經(jīng)過時刻
        properties.put("11", JSONObject.parse("{\"type\":\"date\",\"format\":\"yyyyMMddHHmmss\"}"));// 入庫時間
        properties.put("12", JSONObject.parse("{\"type\":\"keyword\"}"));// 
        clgjxx.put("properties", properties);
        mappings.put("doc", xxxx);
        json.put("mappings", mappings);
        String jsonString = json.toString();
        System.out.println(jsonString);
        HttpEntity entity = new NStringEntity(jsonString, ContentType.APPLICATION_JSON);
        Response response = null;
        try {
            response = restClient.performRequest("PUT", "/_template/123_temp", params, entity);
            System.out.println(
                    "createIndexWithShardNum,response entity is : " + EntityUtils.toString(response.getEntity()));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

新建索引

   public static void add(RestClient restClient, String jsonString) {
        Map<String, String> params = Collections.singletonMap("pretty", "true");
        HttpEntity entity = new NStringEntity(jsonString, ContentType.APPLICATION_JSON);
        Response response = null;
        try {

            response = restClient.performRequest("PUT", "/123_202303", params, entity);
            if (HttpStatus.SC_OK == response.getStatusLine().getStatusCode()
                    || HttpStatus.SC_CREATED == response.getStatusLine().getStatusCode()) {
                // LOG.info("putData,response entity is : " +
                // EntityUtils.toString(response.getEntity()));
                System.out.println(新建ES索引成功！");
            } else {
                System.out.println("新建ES索引失敗！");
                // Assert.fail();
            }
        } catch (Exception e) {
            e.printStackTrace();
            // Assert.fail();
        }
    }

}

批量插入數(shù)據(jù)

1.先批量生成數(shù)據(jù)

JSONObject message = new JSONObject();
            message.put("1", json.get("1"));// 
            //message.put("2", 2);// 
            message.put("3", sf1.format(now));// 入庫時間
            //message.put("4", json.get("4"));// 觸發(fā)時間
            message.put("5", json.get("5"));// 
            message.put("6", json.get("6"));
            //message.put("7", json.get("7"));
            
 JSONObject title = JSONObject.parseObject(
                    "{\"create\":{\"_index\":\"" + index + "\",\"_type\":\"doc\",\"_id\":\"" + rowkey + "\"}}");

            if (message.size() > 0) {
                sb.append(title.toJSONString()).append("\n");
                sb.append(message.toJSONString()).append("\n");
            }
            EsRestUtils.add(restClient, sb.toString());

2批量導(dǎo)入方法

 public static void add(RestClient restClient, String jsonString) {
        Map<String, String> params = Collections.singletonMap("pretty", "true");
        HttpEntity entity = new NStringEntity(jsonString, ContentType.APPLICATION_JSON);
        Response response = null;
        try {

            response = restClient.performRequest("POST", "/doc/_bulk", params, entity);
            if (HttpStatus.SC_OK == response.getStatusLine().getStatusCode()
                    || HttpStatus.SC_CREATED == response.getStatusLine().getStatusCode()) {
                // LOG.info("putData,response entity is : " +
                // EntityUtils.toString(response.getEntity()));
            } else {
                System.out.println("插入ES失??！");
                // Assert.fail();
            }
        } catch (Exception e) {
            e.printStackTrace();
            // Assert.fail();
        }
    }