Java 实现ES批量索引
2016-05-11 17:10
417 查看
项目中有部分数据通过Hadoop离线分析之后,需要将这部分数据的检索功能暴露出去,于是考虑搭建一套简单的索引服务。Lucene比较灵活, 可以自定义比较多的规则,比如分词,排序等,除了索引之外,服务化方面需要写一些代码,本着少开发,易维护的原则,采用了ES来搭建索引服务。ES的REST类型的接口还是比较容易上手的,把数据刷到ES之后,第三方可以直接调用ES的接口来进行检索,目前百万级别的时候性能还是比较OK的。以下是简单的demo, 没有多线程,只有批量建索引。
需要注意的是,需要预先在ES服务器建好索引的schema, 另外,文档有ID的时候最好指定索引ID,这样更新的时候可以指定ID进行更新,有个版本在修改的时候会造成index数量的增长预计就是这个原因。
import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import com.google.common.base.Charsets; import com.google.common.base.Preconditions; import com.google.common.collect.Lists; import com.google.common.io.Files; import com.google.common.io.LineProcessor; import com.zuoye.crm.model.Example; import io.searchbox.client.JestClient; import io.searchbox.client.JestClientFactory; import io.searchbox.client.config.HttpClientConfig; import io.searchbox.core.Bulk; import io.searchbox.core.Index; import io.searchbox.core.Search; import io.searchbox.core.SearchResult; import io.searchbox.indices.CreateIndex; import io.searchbox.indices.DeleteIndex; import org.apache.commons.cli.BasicParser; import org.apache.commons.cli.CommandLine; import org.apache.commons.cli.HelpFormatter; import org.apache.commons.cli.Options; import org.elasticsearch.index.query.QueryBuilder; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; import java.util.List; /** * Created with IntelliJ IDEA. * User: hotallen * Date: 2016/5/11 * Time: 16:50 * To change this template use File | Settings | File Templates. */ public class DemoIndexer { private static final Logger LOGGER = LoggerFactory.getLogger(CrmTeacherIndexer.class); private static final String indexName = "demo_v1"; private static final String type = "main"; private static JestClientFactory factory; private static final int maxRetry = 3; //retry times when failed private List<Example> buffers; private static final int batches = 100; //batch size public DemoIndexer() { String uri = System.getProperty("es.uri"); Preconditions.checkNotNull(uri, "es.uri shold be defined!"); factory = new JestClientFactory(); factory.setHttpClientConfig(new HttpClientConfig .Builder(uri) .multiThreaded(true) .build()); buffers = Lists.newArrayList(); } public void buidIndex() { JestClient client = factory.getObject(); try { client.execute(new CreateIndex.Builder(indexName).build()); } catch (Exception e) { LOGGER.warn("buildIndex error! e = {}", e.getMessage()); } } public void dropIndex() { JestClient client = factory.getObject(); try { client.execute(new DeleteIndex.Builder(indexName).build()); } catch (Exception e) { LOGGER.warn("dropIndex error! e = {}", e.getMessage()); } } public void dropIndex(String indexName, String indexType) { JestClient client = factory.getObject(); try { client.execute(new DeleteIndex.Builder(indexName).type(indexType).build()); } catch (Exception e) { LOGGER.warn("dropIndex error! name = {}, type = {}, e = {}", indexName, e.getMessage()); } } /** * index by single * @param example */ public void indexExample(Example example) { String json = JsonUtil.obj2JsonData(example); JestClient client = factory.getObject(); int retry = 0; while (retry < maxRetry) { try { String tId = example.getTeacherId().toString(); Index index = new Index.Builder(json).index(indexName).type(type).id(tId).build(); client.execute(index); break; } catch (Exception e) { e.printStackTrace(); LOGGER.warn("indexExample exception! e = {}, retry = {}", e.getMessage(), retry); retry++; } } } /** * index by batch * @param examples */ public void indexExamples(List<Example> examples) { Bulk.Builder bulkBuilder = new Bulk.Builder(); for (Example example : examples) { String teacherId = example.getTeacherId().toString(); String json = JsonUtil.obj2JsonData(example); Index index = new Index.Builder(json).index(indexName).type(type).id(teacherId).build(); bulkBuilder.addAction(index); } int retry = 0; while (retry < maxRetry) { try { JestClient client = factory.getObject(); client.execute(bulkBuilder.build()); break; } catch (Exception e) { e.printStackTrace(); LOGGER.warn("indexExample exception! e = {}, retry = {}", e.getMessage(), retry); retry++; } } } /** * index from file * @param path */ public void indexFromFile(String path) { Preconditions.checkNotNull(path, "路径不为空!"); File source = new File(path); try { IndexerLine indexerLine = new IndexerLine(); Files.readLines(source, Charsets.UTF_8, indexerLine); if (!buffers.isEmpty()) { indexExamples(buffers); } LOGGER.info("indexed done! count = {}", indexerLine.getResult()); } catch (Exception e) { e.printStackTrace(); LOGGER.warn("indexFrom File error! e = {}, line", e.getMessage()); } } /** * for test * @param queryBuilder * @return */ public List<string> doSearch(QueryBuilder queryBuilder) { List<string> list = Lists.newArrayList(); SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); searchSourceBuilder.query(queryBuilder); Search search = new Search.Builder(searchSourceBuilder.toString()).addIndex(indexName).addType(type).build(); JestClient client = factory.getObject(); try { SearchResult result = client.execute(search); List<SearchResult.Hit<jsonobject void="">> hits = result.getHits(JSONObject.class); for (SearchResult.Hit hit : hits) { list.add(hit.source.toString()); } } catch (Exception e) {} return list; } class IndexerLine implements LineProcessor<integer> { private int rows = 0; @Override public boolean processLine(String line) throws IOException { rows++; try { Example example = JSON.parseObject(line.trim(), Example.class); if (example != null) { buffers.add(example); } } catch (Exception e) { LOGGER.warn("parse Example error! e = {} , line = {}", e, line); } if (rows % batches == 0) { indexTeachers(buffers); buffers.clear(); } return true; } @Override public Integer getResult() { return rows; } } public static void main(String[] args) { Options opts = new Options(); opts.addOption("h", "help", false, "usage: -f index path"); opts.addOption("f", "file", true, "index file path"); BasicParser parser = new BasicParser(); try { CommandLine cl = parser.parse(opts, args); HelpFormatter formatter = new HelpFormatter(); if (cl.getOptions().length > 0) { if (cl.hasOption("h")) { formatter.printHelp("Options", opts); } else { String file = cl.getOptionValue("f"); DemoIndexer demoIndexer = new DemoIndexer(); demoIndexer.indexFromFile(file); } } else { formatter.printHelp("Options", opts); } } catch (Exception e) { e.printStackTrace(); } } } <
需要注意的是,需要预先在ES服务器建好索引的schema, 另外,文档有ID的时候最好指定索引ID,这样更新的时候可以指定ID进行更新,有个版本在修改的时候会造成index数量的增长预计就是这个原因。
相关文章推荐
- java对世界各个时区(TimeZone)的通用转换处理方法(转载)
- java-注解annotation
- java-模拟tomcat服务器
- java-用HttpURLConnection发送Http请求.
- java-WEB中的监听器Lisener
- Android IPC进程间通讯机制
- Android Native 绘图方法
- Android java 与 javascript互访(相互调用)的方法例子
- 介绍一款信息管理系统的开源框架---jeecg
- 聚类算法之kmeans算法java版本
- java实现 PageRank算法
- PropertyChangeListener简单理解
- c++11 + SDL2 + ffmpeg +OpenAL + java = Android播放器
- 插入排序
- 冒泡排序
- 堆排序
- 快速排序
- 二叉查找树