转载请标明原处：http://blog.csdn.net/hu948162999/article/details/50563110

本来这块业务 是放到SolrCloud上去的 ， 然后 採用solr的facet统计查询，

详细代码參考之前写的文章：http://blog.csdn.net/hu948162999/article/details/50162643

近期遇到SolrCloud 遇到一些问题。。查询db时间过长，SolrCloud的长连接CloudSolrServer老timeout。索引的效率也不够满

意。为了稳定。临时先还原solr单机版本号（上线时，被运维打回来了）。搜索日志就用elasticsearch实时去处理。

大概流程：

基于日志系统ELK 的原型下。參考ELK处理nginx日志文章：http://blog.csdn.net/hu948162999/article/details/50502875

还是用logstash正则去解析搜索日志。搜索日志採用log4j生成。logstash检測到传递给elasticsearch。

log4j:

log4j.appender.E.layout.ConversionPattern= %d|%m%n

logstash配置

新增logstash_search.conf:

input {
file {
type => "searchword"
path => ["/home/work/log/hotword/data"]
}
}
filter {
grok {
match => [
"message", "%{TIMESTAMP_ISO8601:timestamp}\|\{%{GREEDYDATA:kvs}\}"
]
}
kv {
source => "kvs"
field_split => ","
value_split => "="
trimkey => " "
}
date {
match => ["timestamp" , "YYYY-MM-dd HH:mm:ss,SSS"]
}
}
output {
elasticsearch {
hosts => ["host1:9200", "host2:9200", "host3:9200", "host4:9200"]
index => "searchword-%{+YYYY.MM.dd}"
}
}

这里要注意聚合操作的时候。

Logstash 自带有一个优化好的模板。其默认的mapping，string类型都是analyzer。

也就是说，默认分

词是採用单字分词的。

改动默认的logstash mapping模板。參考 http://udn.yyuap.com/doc/logstash-best-practice-cn/output/elasticsearch.html

结构例如以下：

watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQv/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center" alt="">

启动logstash：

nohup bin/logstash -f conf/logstash_search.conf &

运行搜索測试。

能够立即在elasticsearch的插件上看到该搜索行为日志的数据索引。这就是elk的实时性了。

watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQv/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center" alt="">

elasticsearch java端

參考指定mapping和聚合查询代码：

Client client=esobj.getClient();  
    SearchResponse response = client.prepareSearch("searchword\*").setTypes("searchword").addAggregation(AggregationBuilders.terms("hotword").field("keyword")).execute().actionGet();  
    Terms terms = response.getAggregations().get("hotword");

/\*\*  
 \* 初始化索引  
 \* @param client  
 \* @param indexName  
 \* @param indexType  
 \* @param cols  
 \* @return 初始化成功,返回true。否则返回false  
 \* @throws Exception  
 \*/  
public static boolean initIndexMapping(Client client, String indexName, String indexType, List<ColumnInfo> cols) throws Exception {  
    if(StringUtil.isEmpty(indexName) || StringUtil.isEmpty(indexType)) {  
        return false;  
    }  
    indexName = indexName.toLowerCase();  
    indexType = indexType.toLowerCase();  
    //推断索引库是否存在  
    if(indicesExists(client, indexName)) {  
         OpenIndexRequestBuilder openIndexBuilder = new OpenIndexRequestBuilder(client.admin().indices(), OpenIndexAction.INSTANCE);  
         openIndexBuilder.setIndices(indexName).execute().actionGet();  
    }else{  
         //不存在则新建索引库  
         client.admin().indices().prepareCreate(indexName).execute().actionGet();  
    }

    TypesExistsRequest ter = new TypesExistsRequest(new String\[\]{indexName.toLowerCase()}, indexType);  
    boolean typeExists = client.admin().indices().typesExists(ter).actionGet().isExists();  
    //假设 存在 返回!不能覆盖mapping  
    if(typeExists) {  
        return true;  
    }  
    //定义索引字段属性  
    XContentBuilder mapping = jsonBuilder().startObject().startObject(indexType).startObject("properties");  
    for (ColumnInfo col : cols) {  
        String colName = col.getName().toLowerCase().trim();  
        String colType = col.getType().toLowerCase().trim();

        if("string".equals(colType)) {  
            mapping.startObject(colName).field("type", colType).field("store", ""+col.isStore()).field("indexAnalyzer", col.getIndexAnalyzer()).field("searchAnalyzer", col.getSearchAnalyzer()).field("include\_in\_all", col.isStore()).field("boost", col.getBoost()).endObject();  
        }else if("long".equals(colType)) {  
            mapping.startObject(colName).field("type", colType).field("index", "not\_analyzed").field("include\_in\_all", false).endObject();  
        }else if("date".equals(colType)) {  
            mapping.startObject(colName).field("type", colType).field("format", "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd").field("index", "not\_analyzed").field("include\_in\_all", false).endObject();  
        }else {  
            mapping.startObject(colName).field("type", "string").field("index", "not\_analyzed").endObject();  
        }

    }  
    mapping.endObject().endObject().endObject();  
    PutMappingRequest mappingRequest = Requests.putMappingRequest(indexName).type(indexType).source(mapping);  
    PutMappingResponse response = client.admin().indices().putMapping(mappingRequest).actionGet();  
    return response.isAcknowledged();  
}