python 操作 Elasticsearch

安装 Elasticsearch 模块

pip install elasticsearch

添加数据

1
2
3
4
5
6
7
8
9
10
11
from elasticsearch import Elasticsearch

# 默认host为localhost,port为9200.但也可以指定host与port
es = Elasticsearch([{'host': '192.168.10.21', 'port': 9200}])
doc = {
     'author': 'kimchy',
    'text': 'Elasticsearch: cool. bonsai cool.',
     'timestamp': localtime(),
      }
res = es.index(index="test-index", doc_type='tweet', id=1, body=doc)
print(res)

如果创建成功会返回以下结果

1
{'_index': 'test-index', '_type': 'tweet', '_id': '1', '_version': 1, 'result': 'created', '_shards': {'total': 2, 'successful': 2, 'failed': 0}, '_seq_no': 0, '_primary_term': 1}

创建索引

1
es.indices.create(index='irisaa')

删除索引

1
es.indices.create(index='irisaa')

查看集群状态

1
es.cluster.health(wait_for_status='yellow', request_timeout=1)

查询数据

按照 id 来查询数据

1
2
res = es.get(index="test-index", doc_type='tweet', id=1)
print(res['_source'])

操作成功后返回如下结果：

1
{'author': 'kimchy', 'text': 'Elasticsearch: cool. bonsai cool.', 'timestamp': [2018, 10, 12, 14, 9, 44, 4, 285, 0]}

按照DSL语句查询

1
2
res = es.search(index="test-index", body={"query": {"match_all": {}}})
print(res)

操作成功后结果，返回如下结果：

1
{'took': 2, 'timed_out': False, '_shards': {'total': 5, 'successful': 5, 'skipped': 0, 'failed': 0}, 'hits': {'total': 1, 'max_score': 1.0, 'hits': [{'_index': 'test-index', '_type': 'tweet', '_id': '1', '_score': 1.0, '_source': {'author': 'kimchy', 'text': 'Elasticsearch: cool. bonsai cool.', 'timestamp': [2018, 10, 12, 14, 9, 44, 4, 285, 0]}}]}}

更新数据

1
2
3
4
5
6
7
8
es = Elasticsearch()
doc = {
     'author': 'kimchy',
    'text': 'ok.',
     'timestamp': localtime(),
      }
result = es.update(index="test-index", doc_type='tweet', id=1, body=doc)
print(result)

删除数据

1
2
3
4
5
6
7
8

es.delete_by_query(index='twtter',doc_type='_doc',body={
   "query": {
     "match": {
       "message": "some message"
     }
   }
 })

批量化导入es

1
2
3
4
5
6
7
8
9
10
from elasticsearch import helpers
def gendata(index, type, jsons):
    for json in jsons:
        yield {
            "_index": index,
            "_type": type,
            "_source": json,
        }
        
helpers.bulk(es, gendata(index='index-test', jsons))