公司项目要求使用 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
网友评论