ElasticSearch bulk 批量同步数据

公司项目要求使用 ElasticSearch ，本篇介绍一下开发环境的数据导入，并没有使用 ES 的 java api，思路如下：

java 生成 .json 文件
curl 向 ElasticSearch 服务器发送 _bulk 请求，完成批量同步

须知：

create  当文档不存在时创建之
index   创建新文档或替换已有文档
update  局部更新文档
delete  删除一个文档

json 格式如下，每一行 json 结束必须换行

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，由于我并没有那么庞大的数据量，所以在写入的时候并没有分文件。

实践：

1）json 文件生成（PS:基础很烂，IO流不太熟，凑合看吧）

/**
 * ElasticSearch 常量
 * ACTION_* : bulk api json key
 * ES_* : ElasticSearch 中常见属性
 * 
 * @author Taven
 *
 */
public class ESConstant {
    
    /**
     * bulk api json key 当文档不存在时创建之
     */
    public static final String ACTION_CREATE = "create";
    
    /**
     * bulk api json key 创建新文档或替换已有文档
     */
    public static final String ACTION_INDEX = "index";
    
    /**
     * bulk api json key 局部更新文档
     */
    public static final String ACTION_UPDATE = "update";
    
    /**
     * bulk api json key 删除一个文档
     */
    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");

    /**
     * 创建 es_data.json 文件并写入数据
     * 
     * @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();// 创建目录

        try {
            if (!file.exists())
                file.createNewFile();// 创建文件
            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);
        }
        
    }

    /**
     * 将站点数据转换为 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 所需的数据类型
            // { 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 服务器发送请求
windows 如何安装 curl 链接在下面，linux 的同学先自行百度。

# cmd 进入 curl\I386 执行以下命令 ，@后面是你的文件所在位置
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

数据同步成功

学习：
ElasticSearch 权威指南（中文版） https://es.xiaoleilu.com/
环境搭建：
Windows 下安装 ElasticSearch & ElasticSearch head https://www.jianshu.com/p/4467cfe4e651
Windows 环境下 curl 安装和使用 https://blog.csdn.net/qq_21126979/article/details/78690960

ElasticSearch bulk 批量同步数据

须知：

实践：

推荐阅读更多精彩内容