公司項(xiàng)目要求使用 ElasticSearch ，本篇介紹一下開發(fā)環(huán)境的數(shù)據(jù)導(dǎo)入，并沒有使用 ES 的 java api，思路如下：

• java 生成 .json 文件
• curl 向 ElasticSearch 服務(wù)器發(fā)送 _bulk 請(qǐng)求，完成批量同步

須知：

create  當(dāng)文檔不存在時(shí)創(chuàng)建之
index   創(chuàng)建新文檔或替換已有文檔
update  局部更新文檔
delete  刪除一個(gè)文檔

json 格式如下，每一行 json 結(jié)束必須換行

POST /_bulk
{"index":{"_index":"station_index","_type":"station","_id":570}}
{"name":"燕南變電站","id":570,"table":"station"}
{"index":{"_index":"station_index","_type":"station","_id":604}}
{"name":"王石變電站","id":604,"table":"station"}
{"index":{"_index":"station_index","_type":"station","_id":605}}
{"name":"鞍山變電站","id":605,"table":"station"}

官方建議 bulk 批次最好不要超過15MB，由于我并沒有那么龐大的數(shù)據(jù)量，所以在寫入的時(shí)候并沒有分文件。

實(shí)踐：

1）json 文件生成（PS:基礎(chǔ)很爛，IO流不太熟，湊合看吧）

/**
 * ElasticSearch 常量
 * ACTION_* : bulk api json key
 * ES_* : ElasticSearch 中常見屬性
 * 
 * @author Taven
 *
 */
public class ESConstant {
    
    /**
     * bulk api json key 當(dāng)文檔不存在時(shí)創(chuàng)建之
     */
    public static final String ACTION_CREATE = "create";
    
    /**
     * bulk api json key 創(chuàng)建新文檔或替換已有文檔
     */
    public static final String ACTION_INDEX = "index";
    
    /**
     * bulk api json key 局部更新文檔
     */
    public static final String ACTION_UPDATE = "update";
    
    /**
     * bulk api json key 刪除一個(gè)文檔
     */
    public static final String ACTION_DELETE = "delete";
    
    /**
     * ES中的索引
     */
    public static final String ES_INDEX = "_index";
    
    /**
     * ES中的類型
     */
    public static final String ES_TYPE = "_type";
    
    /**
     * ES中的id
     */
    public static final String ES_ID = "_id";
    
}

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.wansidong.communicate.beans.ESConstant;
import com.wansidong.communicate.model.Cable4ES;
import com.wansidong.communicate.model.Station4ES;
import com.wansidong.communicate.model.Transmission4ES;

/**
 * ESHelper 工具類
 * 
 * @author Taven
 *
 */
public class ESHelper {

    private static final Logger logger = LoggerFactory.getLogger(ESHelper.class);
    
    /**
     * 分隔符
     */
    private static String separator = System.getProperty("file.separator");

    /**
     * 創(chuàng)建 es_data.json 文件并寫入數(shù)據(jù)
     * 
     * @param staionList
     * @param cableList
     * @param transList
     */
    public static void writeESJsonData(List<Station4ES> staionList) {
        String tomcatPath = System.getProperty("catalina.home");
        String directoryPath = tomcatPath + separator + "data";// 目錄路徑
        String filePath = tomcatPath + separator + "data" + separator + "es_data.json";// 文件路徑
        File directory = new File(directoryPath);// 目錄File
        File file = new File(filePath);// 文件File
        if (!directory.exists())
            directory.mkdirs();// 創(chuàng)建目錄

        try {
            if (!file.exists())
                file.createNewFile();// 創(chuàng)建文件
            FileWriter writer = new FileWriter(filePath);
            writer.write(parseStation4ES(staionList));
            writer.flush();
            writer.close();
            logger.info("json文件已生成! path:" + filePath);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        
    }

    /**
     * 將站點(diǎn)數(shù)據(jù)轉(zhuǎn)換為 json 字符串
     * 
     * @param staionList
     * @return
     * @throws JsonProcessingException
     */
    private static String parseStation4ES(List<Station4ES> staionList) throws JsonProcessingException {
        Map<String, Object> actionMap = new HashMap<String, Object>();// bulk action
        Map<String, Object> metadataMap = new HashMap<String, Object>();// bulk metadata
        Map<String, Object> bodyMap = new HashMap<String, Object>();// bulk request body
        ObjectMapper mapper = new ObjectMapper();
        StringBuffer stringBuffer = new StringBuffer();

        for (Station4ES station4es : staionList) {
            actionMap.clear();
            metadataMap.clear();
            bodyMap.clear();

            // 封裝 bulk 所需的數(shù)據(jù)類型
            // { action: { metadata }}\n
            // { request body }\n
            metadataMap.put(ESConstant.ES_INDEX, station4es.getTable() + "_index");
            metadataMap.put(ESConstant.ES_TYPE, station4es.getTable());
            metadataMap.put(ESConstant.ES_ID, station4es.getId());
            actionMap.put(ESConstant.ACTION_INDEX, metadataMap);// action
            bodyMap.put("id", station4es.getId());
            bodyMap.put("name", station4es.getName());
            bodyMap.put("table", station4es.getTable());

            stringBuffer.append(mapper.writeValueAsString(actionMap));
            stringBuffer.append(System.getProperty("line.separator"));
            stringBuffer.append(mapper.writeValueAsString(bodyMap));
            stringBuffer.append(System.getProperty("line.separator"));
        }
        return stringBuffer.toString();
    }

}

2）使用 curl 向 ES 服務(wù)器發(fā)送請(qǐng)求
windows 如何安裝 curl 鏈接在下面，linux 的同學(xué)先自行百度。

# cmd 進(jìn)入 curl\I386 執(zhí)行以下命令 ，@后面是你的文件所在位置
curl -l -H "Content-Type:application/json" -H "Accept:applic
ation/json" -XPOST localhost:9200/_bulk?pretty --data-binary @F:\apache-tomcat-8.
0.44\data\es_data.json

學(xué)習(xí)：
ElasticSearch 權(quán)威指南（中文版） https://es.xiaoleilu.com/
環(huán)境搭建：
Windows 下安裝 ElasticSearch & ElasticSearch head http://www.itdecent.cn/p/4467cfe4e651
Windows 環(huán)境下 curl 安裝和使用 https://blog.csdn.net/qq_21126979/article/details/78690960