基于Python操作ElasticSearch

84 阅读 0 评论 56 点赞

我是靠谱客的博主敏感时光，这篇文章主要介绍基于Python操作ElasticSearch，现在分享给大家，希望可以做个参考。

文献一：

环境依赖：

　　Python：2.7
　　ES依赖包：pyelasticsearch
　　ElasticSearch：5.5.1 / 6.0.1
　　操作系统：Windows 10 / CentOS 7

　本文主要就ES基本的CRUD操作做以归纳整理，ES官方对Python的依赖支持有很多，eg：pyelasticsearch、ESClient、elasticutils、pyes、rawes、Surfiki Refine等。博主在工作中只涉及到了pyelasticsearch，所以本文主要就该依赖做说明，其他的依赖包可详见官网。
　pyelasticsearch依赖包的安装命令：pip install elasticsearch

　pyelasticsearch依赖所提供的接口不是很多，下面主要从单一操作和批量操作两大类做以讨论和分析。

单一操作

插入
　　create：必须指定待查询的idnex、type、id和查询体body；缺一不可，否则报错
　　index：相比于create，index的用法就相对灵活很多；id并非是一个必选项，如果指定，则该文档的id就是指定值，若不指定，则系统会自动生成一个全局唯一的id赋给该文档。
eg：

复制代码

1
2
3
body = {"name": 'lucy', 'sex': 'female', 'age': 10}
es = Elasticsearch(['localhost:9200'])
es.index(index='indexName', doc_type='typeName', body, id=None)

删除
　　delete：删除指定index、type、id的文档

复制代码

1
es.delete(index='indexName', doc_type='typeName', id='idValue')

查找
　　get：获取指定index、type、id所对应的文档

复制代码

1
es.get(index='indexName', doc_type='typeName', id='idValue')

更新
　　update：跟新指定index、type、id所对应的文档
　　

复制代码

1
es.update(index='indexName', doc_type='typeName', id='idValue', body={待更新字段})

批量操作

条件查询
　　search：查询满足条件的所有文档，没有id属性，且index，type和body均可为None。
　　body的语法格式必须符合DSL（Domain Specific Language ）格式

复制代码

1
2
3
4
5
query = {'query': {'match_all': {}}}# 查找所有文档
query = {'query': {'term': {'name': 'jack'}}}# 查找名字叫做jack的所有文档
query = {'query': {'range': {'age': {'gt': 11}}}}# 查找年龄大于11的所有文档
allDoc = es.search(index='indexName', doc_type='typeName', body=query)
print allDoc['hits']['hits'][0]# 返回第一个文档的内容

条件删除
　　delete_by_query：删除满足条件的所有数据，查询条件必须符合DLS格式

复制代码

1
2
3
query = {'query': {'match': {'sex': 'famale'}}}# 删除性别为女性的所有文档
query = {'query': {'range': {'age': {'lt': 11}}}}# 删除年龄小于11的所有文档
es.delete_by_query(index='indexName', body=query, doc_type='typeName')

条件更新
　　update_by_query：更新满足条件的所有数据，写法同上删除和查询

批量插入、删除、更新
　　bulk：在这重点和大家聊聊bulk方法，前面的所有方法都很简单，唯独这个bulk在笔者开始接触的时候，花费了不少时间；这个方法可以同时执行多个操作，单只请求一次，从而在批量操作的时候，可以很大程度上减少程序系统开销。此外，bulk不仅可以一次性批量执行插入、或者删除操作，还可以在一次请求中，既可以插入、又可以删除和更新操作。
　　但是需要注意的是，任何一种操作都有固定的文档格式，只有完全符合该格式要求，才可执行成功。废话不多说，直接上代码：

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
 doc = [
{"index": {}},
{'name': 'jackaaa', 'age': 2000, 'sex': 'female', 'address': u'北京'},
{"index": {}},
{'name': 'jackbbb', 'age': 3000, 'sex': 'male', 'address': u'上海'},
{"index": {}},
{'name': 'jackccc', 'age': 4000, 'sex': 'female', 'address': u'广州'},
{"index": {}},
{'name': 'jackddd', 'age': 1000, 'sex': 'male', 'address': u'深圳'},
]
doc = [
{'index': {'_index': 'indexName', '_type': 'typeName', '_id': 'idValue'}}
{'name': 'jack', 'sex': 'male', 'age': 10 }
{'delete': {'_index': 'indexName', '_type': 'typeName', '_id': 'idValue'}}
{"create": {'_index' : 'indexName', "_type" : 'typeName', '_id': 'idValue'}}
{'name': 'lucy', 'sex': 'female', 'age': 20 }
{'update': {'_index': 'indexName', '_type': 'typeName', '_id': 'idValue'}}
{'doc': {'age': '100'}}
]
es.bulk(index='indexName'，
doc_type='typeName', body=doc)

　　通过上面两个例子可以看出，在用bulk在批量操作的时候，对于不同的操作类型，一定要与之对应一个操作头信息（eg：{“index”: {}}， {‘delete’: {…}}， …），否则会报TransportError（400, u’illegal_argument_exception’）的错误。
　　说到这里，在实际过程中，很多时候就会在此处要专门去批凑这样的一个字典数组。假设有如下场景：
　　如果要批量插入一批数据，如上述第一个例子，则在现有数据集的基础上，很容易想到一个解决方法：通过list的奇偶合并的方法快速实现所需要的字典数组。在这推荐一种Python的技巧：[::2]和[1::2]来实现奇偶合并。

文献二：

E lasticsearch是一款分布式搜索引擎，支持在大数据环境中进行实时数据分析。它基于Apache Lucene文本搜索引擎，内部功能通过ReST API暴露给外部。除了通过HTTP直接访问Elasticsearch，还可以通过支持Java、JavaScript、Python及更多语言的客户端库来访问。它也支持集成Apache Hadoop环境。Elasticsearch在有些处理海量数据的公司中已经有所应用，如GitHub、Foursquare和SoundCloud等。

elasticsearch 他对外提供了rest的http的接口，貌似很强大的样子。但是咱们的一些数据平台市场会对于elasticsearch的数据进行分析，尤其是实时分析。当然不能用 http的方式。

下面是http的方式的一个demo:

下面是查询，/ceshi是索引，rui是type，搜索的内容是title为jones的数据

1	`curl http://vim.xiaorui.cc:9200/ceshi/rui/_search?q=title:jones&size=5&pretty=true`

添加数据

1	`curl` `-X POST` `-d` `'{"title":"jones","amount":5.7}'`

但是听说，1.x之后不能直接curl，这不是重点忽略下面介绍一个python使用elasticsearch的例子

复制代码

from datetime import datetime
from elasticsearch import Elasticsearch
#连接elasticsearch,默认是9200
es = Elasticsearch()
#创建索引，索引的名字是my-index,如果已经存在了，就返回个400，
#这个索引可以现在创建，也可以在后面插入数据的时候再临时创建
es.indices.create(index='my-index',ignore)
#{u'acknowledged':True}
#插入数据,(这里省略插入其他两条数据，后面用)
es.index(index="my-index",doc_type="test-type",id=01,body={"any":"data01","timestamp":datetime.now()})
#{u'_type':u'test-type',u'created':True,u'_shards':{u'successful':1,u'failed':0,u'total':2},u'_version':1,u'_index':u'my-index',u'_id':u'1}
#也可以，在插入数据的时候再创建索引test-index
es.index(index="test-index",doc_type="test-type",id=42,body={"any":"data","timestamp":datetime.now()})
#查询数据，两种get and search
#get获取
res = es.get(index="my-index", doc_type="test-type", id=01)
print(res)
#{u'_type': u'test-type', u'_source': {u'timestamp': u'2016-01-20T10:53:36.997000', u'any': u'data01'}, u'_index': u'my-index', u'_version': 1, u'found': True, u'_id': u'1'}
print(res['_source'])
#{u'timestamp': u'2016-01-20T10:53:36.997000', u'any': u'data01'}
#search获取
res = es.search(index="test-index", body={"query":{"match_all":{}}})
print(res)
#{u'hits':
#    {
#    u'hits': [
#        {u'_score': 1.0, u'_type': u'test-type', u'_id': u'2', u'_source': {u'timestamp': u'2016-01-20T10:53:58.562000', u'any': u'data02'}, u'_index': u'my-index'},
#        {u'_score': 1.0, u'_type': u'test-type', u'_id': u'1', u'_source': {u'timestamp': u'2016-01-20T10:53:36.997000', u'any': u'data01'}, u'_index': u'my-index'},
#        {u'_score': 1.0, u'_type': u'test-type', u'_id': u'3', u'_source': {u'timestamp': u'2016-01-20T11:09:19.403000', u'any': u'data033'}, u'_index': u'my-index'}
#    ],
#    u'total': 5,
#    u'max_score': 1.0
#    },
#u'_shards': {u'successful': 5, u'failed': 0, u'total':5},
#u'took': 1,
#u'timed_out': False
#}
for hit in res['hits']['hits']:
    print(hit["_source"])
res = es.search(index="test-index", body={'query':{'match':{'any':'data'}}}) #获取any=data的所有值
print(res)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

from datetime import datetime
from elasticsearch import Elasticsearch
#连接elasticsearch,默认是9200
es = Elasticsearch()
#创建索引，索引的名字是my-index,如果已经存在了，就返回个400，
#这个索引可以现在创建，也可以在后面插入数据的时候再临时创建
es.indices.create(index='my-index',ignore)
#{u'acknowledged':True}
#插入数据,(这里省略插入其他两条数据，后面用)
es.index(index="my-index",doc_type="test-type",id=01,body={"any":"data01","timestamp":datetime.now()})
#{u'_type':u'test-type',u'created':True,u'_shards':{u'successful':1,u'failed':0,u'total':2},u'_version':1,u'_index':u'my-index',u'_id':u'1}
#也可以，在插入数据的时候再创建索引test-index
es.index(index="test-index",doc_type="test-type",id=42,body={"any":"data","timestamp":datetime.now()})
#查询数据，两种get and search
#get获取
res = es.get(index="my-index", doc_type="test-type", id=01)
print(res)
#{u'_type': u'test-type', u'_source': {u'timestamp': u'2016-01-20T10:53:36.997000', u'any': u'data01'}, u'_index': u'my-index', u'_version': 1, u'found': True, u'_id': u'1'}
print(res['_source'])
#{u'timestamp': u'2016-01-20T10:53:36.997000', u'any': u'data01'}
#search获取
res = es.search(index="test-index", body={"query":{"match_all":{}}})
print(res)
#{u'hits':
#    {
#    u'hits': [
#        {u'_score': 1.0, u'_type': u'test-type', u'_id': u'2', u'_source': {u'timestamp': u'2016-01-20T10:53:58.562000', u'any': u'data02'}, u'_index': u'my-index'},
#        {u'_score': 1.0, u'_type': u'test-type', u'_id': u'1', u'_source': {u'timestamp': u'2016-01-20T10:53:36.997000', u'any': u'data01'}, u'_index': u'my-index'},
#        {u'_score': 1.0, u'_type': u'test-type', u'_id': u'3', u'_source': {u'timestamp': u'2016-01-20T11:09:19.403000', u'any': u'data033'}, u'_index': u'my-index'}
#    ],
#    u'total': 5,
#    u'max_score': 1.0
#    },
#u'_shards': {u'successful': 5, u'failed': 0, u'total':5},
#u'took': 1,
#u'timed_out': False
#}
for hit in res['hits']['hits']:
    print(hit["_source"])
res = es.search(index="test-index", body={'query':{'match':{'any':'data'}}}) #获取any=data的所有值
print(res)