最近我们需要对大约2T（6.5亿条）日志做全文检索，Elasticsearch看起来很火爆，又有很多产品使用（Facebook、github、stackoverflow），值得一试。以下是一些基础知识笔记。
Elasticsearch是一个基于Lucene构建的开源、分布式、RESTful的搜索引擎，能够实现近实时（NRT）搜索，稳定、可靠、安装方便。性能不错、水平扩展、文档齐全、社区火爆，这几点很重要。
如果之前已经了解过分布式存储系统、query切词、检索相关性策略，Elasticsearch的使用学习起来会很快。
1 基础概念
Elasticsearch是一个近实时的系统，从你写入数据到数据可以被检索到，一般会有1秒钟的延时。Elasticsearch是基于Lucene的，Lucene的读写是两个分开的句柄，往写句柄写入的数据刷新之后，读句柄重新打开，这才能读到新写入的数据。
名词解释：
Cluster：集群。
Index：索引，Index相当于关系型数据库的DataBase。
Type：类型，这是索引下的逻辑划分，一般把有共性的文档放到一个类型里面，相当于关系型数据库的table。
Document：文档，Json结构，这点跟MongoDB差不多。
Shard、Replica：分片，副本。
分片有两个好处，一个是可以水平扩展，另一个是可以并发提高性能。在网络环境下，可能会有各种导致分片无法正常工作的问题，所以需要有失败预案。ES支持把分片拷贝出一份或者多份，称为副本分片，简称副本。副本有两个好处，一个是实现高可用（HA，High Availability），另一个是利用副本提高并发检索性能。
分片和副本的数量可以在创建index的时候指定，index创建之后，只能修改副本数量，不能修改分片。
健康状态：
安装了head插件之后，可以在web上看到集群健康状态，集群处于绿色表示当前一切正常，集群处于黄色表示当前有些副本不正常，集群处于红色表示部分数据无法正常提供。绿色和黄色状态下，集群都是能提供完整数据的，红色状态下集群提供的数据是有缺失的。
2 搭建ElasticSearch
首先安装java，设置好JAVA_HOME环境变量（export JAVA_HOME=.../java8），然后安装Elasticsearch。
参考官方文档：https://www.elastic.co/guide/en/elasticsearch/reference/current/_installation.html
设置配置的时候，ES可能因为各种原因不能自动找到集群，所以把地址也设置上，如：
discovery.zen.ping.unicast.hosts: ["host_name...:9301", "host_name_xxx:port_yyy"...]
安装head插件:拉取 https://github.com/mobz/elasticsearch-head 代码，将其放到./plugins/head 目录下。
启动之前设置ES使用的内存：export ES_HEAP_SIZE=10g。
elasticsearcy.yml配置文件中的一些配置点：
#设置集群名字
cluster.name: cswuyg_qa_pair_test
#设置node名字
node.name: xxx-node
#设置节点域名
network.host: 10.111.111.1
#设置内部传输端口和外部HTTP访问端口
transport.tcp.port: 9302
http.port: 8302
#设置集群其它节点地址
discovery.zen.ping.unicast.hosts: ["xxxhost:yyyport"]
#设置中文切词插件
index.analysis.analyzer.ik.type: "ik"
elasticsearch -d 以守护进程方式启动，启动之后，就可以在浏览器里使用head插件看到集群信息，如：
http://host_name_xxx:port_yyy/_plugin/head/
上图：启动了三个Elasticsearch实例，创建了三个Index；ceshi Index有一主shard，两replica shard；qa_pair1 Index只有主shard；website Index有一主shard，一replica shard。
3 测试Elasticsearch使用
Elasticsearch提供RESTful API，我采用Postman（chrome的一个插件）作为辅助客户端向ES发送请求。
可以向任意一个节点发起请求，虽然ES有Master的概念，但任意一个node都可以接受读写请求。
先创建一个index：
POST http://10.11.111.11:8301/test_index
查看创建的index:
GET http://10.11.111.11:8301/_cat/indices?v
写入数据：
查询数据：
（1）使用id直接查：
GET http://xxxhost:8201/qa_xx2/qa_xx3/1235
（2）DSL查询：
往查询url POST数据即可：
URL格式：http://xxxhost:8201/qa_xx2/qa_xx3/_search
a. 查询title中包含有cswuyg字段的文档。Highlight设置高亮命中的词。POST方法的body：
{ "query": { "match": { "title": { "query": "cswuyg " } } }, "highlight": { "fields": { "title": { } } } }
b. bool组合查询，命中的文档的title字段必须能命中“餐厅”、“好吃”、“深圳”，可以是完全命中，也可以是名字其中的个别字。“便宜”则是可选命中。
POST方法的body：
{ "query": { "bool": { "must": [{ "match": { "title": { "query": "餐厅" } } }, { "match": { "title": { "query": "好吃" } } }, { "match": { "title": { "query": "深圳" } } }], "should": [{ "match": { "title": "便宜" } }] } }, "highlight": { "fields": { "title": { } } } }
如果要求每一个字都命中，可以把match修改为match_phrase。
{ 'query': { 'bool': { 'should': [{ 'match': { 'title': { 'query': '张三', 'boost': 0.2 } } }], 'must': [{ 'match_phrase': { 'title': { 'query': '李四', 'boost': 0.69 } } }, { 'match_phrase': { 'title': { 'query': '王五', 'boost': 0.11 } } }] } } }
例子：要求必须完全命中“酒后”和“标准"，“驾驶”可以部分命中
{ "query": { "bool": { "must": [{ "match_phrase": { "question": { "query": "酒后", "boost": 0.69 } } }, { "match": { "question": { "query": "驾驶", "boost": 0.11 } } }, { "match_phrase": { "question": { "query": "标准", "boost": 0.2 } } }] } } }
c. 给查询词设置权重（boost）。POST方法的body：
{ "query": { "bool": { "must": { "match": { "title": { "query": "好吃的餐厅", "boost": 1 } } }, "must": { "match": { "title": { "query": "深圳湾", "boost": 100 } } }, "should": [{ "match": { "title": "便宜" } }] } }, "highlight": { "fields": { "title": { } } } }
d. filter查询，也就是kv查询，不涉及检索的相关性打分，title必须是完全命中，如果建库时是有对这个字段切词的，则查询时，需要是切词后的某个词去查询，如“今天天气”，建库切词为“今天”和“天气”，那么filter查询的时候需要使用“今天”或者“天气”才能命中。POST方法的body：
{ "query": { "bool": { "filter": [{ "term": { "title": "好吃的" } }] } } }
e. 完全匹配某个短语,这就要求“好厉害”三个字组成的词必须在文档中出现，不能是只出现其中的个别字（match就是这样）。POST方法的body：
{ "query": { "match_phrase": { "title": { "query": "好厉害" } } } }
（3）运维
a. 去掉副本，调研的时候希望不要副本，这样子写入会快点
PUT http://10.11.111.11:8202/qa_pair2/_settings { "number_of_replicas" : 0 }
4 使用ik中文切词插件
Elasticsearch默认的中文切词插件是单字切词，这不能满足我们要求，需要安装中文切词插件。
插件github地址：https://github.com/medcl/elasticsearch-analysis-ik
源码安装：编译时需要联网，可以在windows下编译完之后，把elasticsearch-analysis-ik-1.9.3.zip拷贝到linux机器的./plugin/head目录下解压。
配置：在配置文件./config/elasticsearch.yml末尾添加配置： index.analysis.analyzer.ik.type: "ik"
测试ik切词：http://host_name_xx:port_yyy/qa_pair/_analyze?analyzer=ik&pretty=true&text=我是中国人"
5 使用python读写Elasticsearch
驱动安装：使用pip安装elasticsearch
读取文件批量插入数据示例：
#!/home/work/bin/python #-*-coding:utf8-*- """ 读取文件，入库到es 使用：python insert_demo.py xxx_file_name Authors: cswuyg Date: 2016.06.18 """ from elasticsearch import Elasticsearch from elasticsearch.helpers import bulk from elasticsearch import exceptions import traceback import datetime import sys reload(sys) sys.setdefaultencoding('utf-8') #设置mappings def _create_index(es, index_name="cswuyg", doc_type_name="cswuyg"): my_settingss = { 'number_of_shards': 18, 'number_of_replicas': 0 } my_mappings = { "cswuyg": { '_all': { 'enabled': 'false' }, "properties": { "title": { 'type': 'string', 'store': 'no', 'term_vector': 'with_positions_offsets', 'analyzer': 'ik_max_word', 'boost': 8 }, "url": { "type": "string", 'index': 'not_analyzed' }, 'content': { 'type': 'string', 'store': 'no', 'term_vector': 'with_positions_offsets', 'analyzer': 'ik_max_word', 'boost': 8 } } } } settings = { 'settings': my_settingss, 'mappings': my_mappings } create_index = es.indices.create(index=index_name, body=settings) #将文件中的数据存储到es中 def _save_data(es, input_file): #读入数据 all_data = list() count = 0 with open(input_file) as f_r: for line in f_r: count += 1 all_data.append({ '_index': 'cswuyg', '_type': 'cswuyg', '_source': { 'title': line } }) if len(all_data) == 100: success, _ = bulk(es, all_data, index='cswuyg', raise_on_error=True) all_data = list() print('{1}: finish {0}'.format(count, input_file)) if len(all_data) != 0: success, _ = bulk(es, all_data, index='cswuyg', raise_on_error=True) all_data = list() print('{1}: finish {0}'.format(count, input_file)) print('{0}: finish all'.format(input_file)) def _insert_data(es, file_name): start_time = datetime.datetime.now() _save_data(es, file_name) cost_time = datetime.datetime.now() - start_time print('all cost time{0}'.format(cost_time)) def _main(): if len(sys.argv) != 2: print('need file argument') return es = Elasticsearch(hosts=["10.200.100.80:8301"], timeout=500) try: _create_index(es) except exceptions.RequestError: print(traceback.format_exc()) _insert_data(es, sys.argv[1]); if __name__ == '__main__': _main()
测试文件数据格式：
我是中国人
我爱中国
检索示例（从文件中获取检索条件：切好词且打好分的Term）：
#!/home/work/bin/python #-*-coding:utf8-*- """ 检索 es 策略： 从文件中读取已经切好词，且给好权重的term，且选好必选term的一行文本，处理成match_phrase + bool查询 默认策略2 使用方法：python search_demo.py test_file output: es_query query title sall_score output demo: {'query': {'bool': {'should': [{'match': {'title': {'query': 'xe6x88x91', 'boost': 0.2}}}], 'must': [{'match_phrase': {'title': {'query': 'xe4xb8xadxe5x9bxbd', 'boost': 0.69}}}, {'match_phrase': {'title': {'query': 'xe7x88xb1', 'boost': 0.11}}}]}}} 我爱中国 我爱中国 {"should": ["我"], "score": {"我": 0.2, "中国": 0.69, "爱": 0.11}, "must": ["中国", "爱"]} 我爱中国 我爱中国 {"should": ["我"], "score": {"我": 0.2, "中国": 0.69, "爱": 0.11}, "must": ["中国", "爱"]} Authors: cswuyg Date: 2016.06.18 """ from elasticsearch import Elasticsearch from elasticsearch.helpers import bulk from elasticsearch import exceptions import sys import json from log import logger import traceback reload(sys) sys.setdefaultencoding('utf-8') def _release_es_query_by_file_info(file_info): #输入格式： raw_query must_term should_term all_score logger.debug('file_info:{0}'.format(file_info)) file_info_list = file_info.split(' ') print file_info_list raw_query = file_info_list[0] must_term = file_info_list[3] should_term= file_info_list[4] all_score = file_info_list[5] json_score = json.loads(all_score, encoding='utf-8') ret_obj = {} ret_obj['must'] = must_term.split() ret_obj['should'] = should_term.split() ret_obj['score'] = json_score bool_query = dict() must_query = list() should_query = list() for item in must_term.split(' '): must_query.append({'match_phrase': {'title': {'query': item, 'boost': json_score[unicode(item)]}}}) bool_query['must'] = must_query for item in should_term.split(' '): should_query.append({'match': {'title': {'query': item, 'boost': json_score[unicode(item)]}}}) bool_query['should'] = should_query es_query = {'query': {'bool': bool_query}} print es_query return raw_query, es_query, json.dumps(ret_obj, ensure_ascii=False) def _do_query_use_file_info(es, file_info): raw_query, query, all_score = _release_es_query_by_file_info(file_info.strip(' ')) res = es.search(index='cswuyg', doc_type='cswuyg', body=query, size=100) if (len(res['hits']['hits']) == 0): logger.debug('len(res["hits"]["hits"]) == 0') print("{0} {1} {2}".format(raw_query, "", all_score)) return for item in res['hits']['hits']: try: print("{0} {1} {2}".format(raw_query, item['_source']['title'].strip(' '), all_score)) except: logger.debug(traceback.format_exc()) logger.debug(item['_source']['title']) print(' ') def _main(): if len(sys.argv) != 2: print('argv error') return else: print('argv[1] = {0}'.format(sys.argv[1])) es = Elasticsearch(hosts=["10.200.100.80:8301"], timeout=5000) with open(sys.argv[1]) as f_r: for item in f_r: try: _do_query_use_file_info(es, item) except: logger.debug(traceback.format_exc()) if __name__ == '__main__': _main()
测试文件数据格式：
我爱中国 中国 爱 我 {"我": 0.20, "中国": 0.69, "爱": 0.11}
6 部分学习资料
官方入门文档：https://www.elastic.co/guide/en/elasticsearch/reference/current/getting-started.html
使用ik中文分词插件：http://blog.csdn.net/liuzhenfeng/article/details/39404435
提升性能：http://zhousheng29.iteye.com/blog/2101905
Elasticsearch有哪些用户：https://www.elastic.co/use-cases
附上我们的ES使用效果：
使用了ES 2.3版本，集群使用了4台机器18块磁盘，启了18个节点，每个节点15G内存，共270G内存。Index无副本，disable掉了_all索引，2T数据入库完占了大约4T磁盘空间，10进程并发写入，速度可以达到1W条/s（写入多了后面会变慢），部分磁盘写入延迟达到几百毫秒，瓶颈在磁盘IO上。首次召回100W+的文档耗时2s+，但如果是触发了缓存的召回，可以达到50ms级别的耗时。多加机器，增加shard可以提高读写性能。
后续如有继续使用再持续更新。