APDPlat拓展搜索之集成ElasticSearch

2019-03-27 01:23|来源: 网路

APDPlat充分利用Compass的OSEMORM integration特性,提供了简单易用功能强大内置搜索特性。

 

APDPlat的内置搜索,在设计简洁优雅的同时,还具备了强大的实时搜索能力,用户只需用注解的方式在模型中指定需要搜索哪些字段(还可在模型之间进行关联搜索)就获得了搜索能力,而不用编写任何代码。平台自动处理索引维护、查询解析、结果高亮等支撑功能。

 

然而APDPlat的内置搜索只能在单机上面使用,不支持分布式,只能用于中小规模的场景。为了支持大规模的分布式搜索和实时分析,APDPlat选用Compass的进化版ElasticSearch Compass和ElasticSearch的关系)。

 

ElasticSearch提供了Java Client API,但是由于该API依赖于Lucene的org.apache.lucene.util包中的几个类,以致于无法和APDPlat集成,原因是APDPlat中Compass依赖的Lucene的版本和ElasticSearch依赖的版本冲突。

 

从这里可以得知,ElasticSearch的Java Client API如果完全移除对Lucene的依赖,仅仅作为用户和ElasticSearch集群之间通信的接口,使用起来就会更方便。

 

因此,APDPlat只能采用ElasticSearch的RESTful API

 

接下来我们看一个APDPlat和ElasticSearch集成的例子:

 

APDPlat提供了可扩展的日志处理接口,用户可编写自己的插件并在配置文件中指定启用哪些插件,日志处理接口如下:

 

/**
 * 日志处理接口:
 * 可将日志存入独立日志数据库(非业务数据库)
 * 可将日志传递到activemq\rabbitmq\zeromq等消息队列
 * 可将日志传递到kafka\flume\chukwa\scribe等日志聚合系统
 * 可将日志传递到elasticsearch\solr等搜索服务器
 * @author 杨尚川
 */
public interface LogHandler {
    public <T extends Model> void handle(List<T> list);
}

 

将日志传递到ElasticSearch搜索服务器的实现使用了几个配置信息,这些配置信息默认存放在config.properties中,如下所示:

 

#elasticsearch服务器配置
elasticsearch.host=localhost
elasticsearch.port=9200
elasticsearch.log.index.name=apdplat_for_log

 

因为LogHandler接口中定义的参数List<T> list为泛型,只知道T是Model的子类,而不知道具体是哪一个类,所以我们使用反射的机制来获取具体对象类型:

 

String simpleName = model.getClass().getSimpleName();
LOG.debug((j++)+"、simpleName: 【"+simpleName+"】");
json.append("{\"index\":{\"_index\":\"")
  .append(INDEX_NAME)
  .append("\",\"_type\":\"")
  .append(simpleName)
  .append("\"}}")
  .append("\n");
json.append("{");

 

同时,我们利用反射的方式获取对象的字段以及相应的值,并正确处理类型问题:

 

Field[] fields = model.getClass().getDeclaredFields();
int len = fields.length;
for(int i = 0; i < len; i++){
	Field field = fields[i];
	String name = field.getName();
	field.setAccessible(true);
	Object value = field.get(model);
	//小心空指针异常,LogHandler线程会悄无声息地退出!
	if(value == null){
		LOG.debug("忽略空字段:"+name);
		continue;
	}
	if(i>0){
		json.append(",");
	}
	String valueClass=value.getClass().getSimpleName();
	LOG.debug("name: "+name+"   type: "+valueClass);
	if("Timestamp".equals(valueClass) || "Date".equals(valueClass)){
		//提交给ES的日期时间值要为"2014-01-31T13:53:54"这样的形式
		value=DateTypeConverter.toDefaultDateTime((Date)value).replace(" ", "T");
	}
	String prefix = "\"";
	String suffix = "\"";
	//提交给ES的数字和布尔值不要加双引号
	if("Float".equals(valueClass)
			|| "Double".equals(valueClass) 
			|| "Long".equals(valueClass) 
			|| "Integer".equals(valueClass)
			|| "Short".equals(valueClass)
			|| "Boolean".equals(valueClass)){
		prefix="";
		suffix="";
	}
	json.append("\"")
			.append(name)
			.append("\":")
			.append(prefix)
			.append(value)
			.append(suffix);
}
json.append("}\n");

 

构造完要提交的JSON数据之后,向服务器发送HTTP PUT请求:

 

HttpURLConnection conn = (HttpURLConnection) URL.openConnection();
conn.setRequestMethod("PUT");
conn.setDoOutput(true);
BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(conn.getOutputStream(),"utf-8"));    
writer.write(json.toString());
writer.flush();
StringBuilder result = new StringBuilder();
try (BufferedReader reader = new BufferedReader (new InputStreamReader (conn.getInputStream()))) {
	String line = reader.readLine();
	while(line != null){
		result.append(line);
		line = reader.readLine();
	}
}

 

服务器会以JSON数据格式返回处理结果,我们使用Jackson解析返回的JSON字符串:

 

JsonNode node = MAPPER.readTree(resultStr);
for(JsonNode item : node.get("items")){
	JsonNode createJsonNode = item.get("create");
	JsonNode okJsonNode = createJsonNode.get("ok");
	if(okJsonNode != null){
		boolean r = okJsonNode.getBooleanValue();
		if(r){
			success++;
		}
	}else{
		JsonNode errorJsonNode = createJsonNode.get("error");
		if(errorJsonNode != null){
			String errorMessage = errorJsonNode.getTextValue();
			LOG.error("索引失败:"+errorMessage);
		}
	}
}

 

下面是ElasticSearchLogHandler完整的实现:

 

/**
 * 
 * 日志处理实现:
 * 将日志保存到ElasticSearch中
 * 进行高性能实时搜索和分析
 * 支持大规模分布式搜索
 * 
 * @author 杨尚川
 */
@Service
public class ElasticSearchLogHandler implements LogHandler{
    private static final APDPlatLogger LOG = new APDPlatLogger(ElasticSearchLogHandler.class);
    
    private static final String INDEX_NAME = PropertyHolder.getProperty("elasticsearch.log.index.name");
    private static final String HOST = PropertyHolder.getProperty("elasticsearch.host");
    private static final String PORT = PropertyHolder.getProperty("elasticsearch.port");
    private static final ObjectMapper MAPPER = new ObjectMapper();
    private static URL URL;
    
    private int success;
   
    public ElasticSearchLogHandler(){
        LOG.info("elasticsearch.log.index.name: "+INDEX_NAME);
        LOG.info("elasticsearch.host: "+HOST);
        LOG.info("elasticsearch.port: "+PORT);
        try {
            URL = new URL("http://"+HOST+":"+PORT+"/_bulk");
        } catch (MalformedURLException ex) {
            LOG.error("构造URL失败",ex);
        }
    }
    
    /**
     * 批量索引
     * 批量提交
     * 
     * @param <T> 泛型参数
     * @param list 批量模型
     */
    public <T extends Model> void index(List<T> list){
        success = 0;
        StringBuilder json = new StringBuilder();
        int j = 1;
        //构造批量索引请求
        for(T model : list){
            try{
                String simpleName = model.getClass().getSimpleName();
                LOG.debug((j++)+"、simpleName: 【"+simpleName+"】");
                json.append("{\"index\":{\"_index\":\"")
                        .append(INDEX_NAME)
                        .append("\",\"_type\":\"")
                        .append(simpleName)
                        .append("\"}}")
                        .append("\n");
                json.append("{");
                Field[] fields = model.getClass().getDeclaredFields();
                int len = fields.length;
                for(int i = 0; i < len; i++){
                    Field field = fields[i];
                    String name = field.getName();
                    field.setAccessible(true);
                    Object value = field.get(model);
                    //小心空指针异常,LogHandler线程会悄无声息地退出!
                    if(value == null){
                        LOG.debug("忽略空字段:"+name);
                        continue;
                    }
                    if(i>0){
                        json.append(",");
                    }
                    String valueClass=value.getClass().getSimpleName();
                    LOG.debug("name: "+name+"   type: "+valueClass);
                    if("Timestamp".equals(valueClass) || "Date".equals(valueClass)){
                        //提交给ES的日期时间值要为"2014-01-31T13:53:54"这样的形式
                        value=DateTypeConverter.toDefaultDateTime((Date)value).replace(" ", "T");
                    }
                    String prefix = "\"";
                    String suffix = "\"";
                    //提交给ES的数字和布尔值不要加双引号
                    if("Float".equals(valueClass)
                            || "Double".equals(valueClass) 
                            || "Long".equals(valueClass) 
                            || "Integer".equals(valueClass)
                            || "Short".equals(valueClass)
                            || "Boolean".equals(valueClass)){
                        prefix="";
                        suffix="";
                    }
                    json.append("\"")
                            .append(name)
                            .append("\":")
                            .append(prefix)
                            .append(value)
                            .append(suffix);
                }
                json.append("}\n");
            }catch(SecurityException | IllegalArgumentException | IllegalAccessException e){
                LOG.error("构造索引请求失败【"+model.getMetaData()+"】\n"+model, e);
            }
        }
        //批量提交索引
        try{
            LOG.debug("提交JSON数据:\n"+json.toString());
            HttpURLConnection conn = (HttpURLConnection) URL.openConnection();
            conn.setRequestMethod("PUT");
            conn.setDoOutput(true);
            BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(conn.getOutputStream(),"utf-8"));    
            writer.write(json.toString());
            writer.flush();
            StringBuilder result = new StringBuilder();
            try (BufferedReader reader = new BufferedReader (new InputStreamReader (conn.getInputStream()))) {
                String line = reader.readLine();
                while(line != null){
                    result.append(line);
                    line = reader.readLine();
                }
            }
            String resultStr = result.toString();
            LOG.debug(resultStr);          
            //使用Jackson解析返回的JSON
            JsonNode node = MAPPER.readTree(resultStr);
            for(JsonNode item : node.get("items")){
                JsonNode createJsonNode = item.get("create");
                JsonNode okJsonNode = createJsonNode.get("ok");
                if(okJsonNode != null){
                    boolean r = okJsonNode.getBooleanValue();
                    if(r){
                        success++;
                    }
                }else{
                    JsonNode errorJsonNode = createJsonNode.get("error");
                    if(errorJsonNode != null){
                        String errorMessage = errorJsonNode.getTextValue();
                        LOG.error("索引失败:"+errorMessage);
                    }
                }
            }
        }catch(IOException e){
            LOG.error("批量提交索引失败", e);
        }
    }
    
    @Override
    public <T extends Model> void handle(List<T> list) {
        LOG.info("开始将 "+list.size()+" 个日志对象索引到ElasticSearch服务器");
        long start = System.currentTimeMillis();
        index(list);
        long cost = System.currentTimeMillis() - start;
        
        if(success != list.size()){
            LOG.info("索引失败: "+(list.size()-success)+" 个");            
        }
        if(success > 0){
            LOG.info("索引成功: "+success+" 个");
        }
        LOG.info("耗时:"+ConvertUtils.getTimeDes(cost));
    }
}

 

最后我们在配置文件config.local.properties中指定log.handlers的值为ElasticSearchLogHandler类的Spring bean name elasticSearchLogHandler,因为ElasticSearchLogHandler类加了Spring的@Service注解:

 

log.handlers=elasticSearchLogHandler

 

 

APDPlat托管在Github


转自:http://my.oschina.net/apdplat/blog/197012

相关问答

更多

关于elasticsearch和hadoop集成该怎么做的搜索推荐

配置elasticsearch的存储路径为hdfs需要两步,安装插件elasticsearch-hadoop,在联网的情况下在命令窗口运行:plugin -install elasticsearch/elasticsearch-hadoop/1.2.0即可。 如果没有联网解压插件到plugins中即可,目录为/hadoop。。。。。 ...

保护弹性搜索(Securing elasticsearch)

有一篇关于保护Elasticsearch的文章,其中包含了很多要注意的要点: http : //www.found.no/foundation/elasticsearch-security/ (完全披露:我写了它并为Found工作) 你还应该知道一些事情: http : //www.found.no/foundation/elasticsearch-in-production/ 总结一下总结: 目前,Elasticsearch并不认为安全性是其工作。 Elasticsearch没有用户的概念。 从 ...

Elasticsearch:从点击中学习(搜索结果排名)(Elasticsearch: Learning from clicks (Search result ranking))

在信息检索领域(搜索和推荐的一般学术领域),这通常被称为学习排名 。 无论是点击次数,转化次数还是其他形式的搜索关键字搜索的“好”或“坏”结果,学习排名都会使用分类器或回归过程来了解查询和文档的哪些功能与相关性相关联。 点击? 对于具体的点击,有理由怀疑优化点击是理想的。 微软研究院的一篇论文我试图挖掘出这样的说法,即在他们的情况下,点击率与相关性只有45%相关。 Click + dwell通常是一个更有用的通用相关指标。 正如我在这篇博客文章中所谈到的那样,搜索中存在自我强化偏见的风险。 如果

Elasticsearch - 升级到2.1.1后搜索失败(Elasticsearch - Search fails after upgrading to 2.1.1)

我通过着色elasticsearch依赖来解决这个问题 1.我创建了一个单独的maven模块: <project> ... <dependencies> <dependency> <groupId>org.elasticsearch</groupId> <artifactId>elasticsearch</artifactId> <version>2.3.4</version> </dependency> </dependencie

elasticsearch怎么文件搜索

Elasticsearch是一个分布式的文档(document)存储引擎。它可以实时存储并检索复杂数据结构——序列化的JSON文档。换言说,一旦文档被存储在Elasticsearch中,它就可以在集群的任一节点上被检索。 当然,我们不仅需要存储数据,还要快速的批量查询。虽然已经有很多NoSQL的解决方案允许我们以文档的形式存储对象,但它们依旧需要考虑如何查询这些数据,以及哪些字段需要被索引以便检索时更加快速。 程序中大多的实体或对象能够被序列化为包含键值对的JSON对象,键(key)是字段(fie ...