python中操作es

共计 6086 个字符，预计需要花费 16 分钟才能阅读完成。

最近内部用了es，博主呢也乘机再项目前期试用下。。。。指不定哪天项目取消了就销毁资源了

安装依赖包

pip3 install elasticsearch -i https://pypi.douban.com/simple

测试连接

初始es链接

import time
from elasticsearch import Elasticsearch

es = Elasticsearch(
    hosts=["192.168.44.142:9200","192.168.44.143:9200","192.168.44.144:9200",],
    http_auth=('elastic','elastic'),
)

print(es.ping())

索引操作

# 创建索引
print(es.indices.create(index="test1"))
# 查看索引
print(es.cat.indices())
# 删除索引
print(es.indices.delete("test1"))

若索引存在则忽略返回的400报错

# 方式一
from elasticsearch.exceptions import RequestError
try:
    es.indices.create(index='my-index')
except RequestError as e:
    print('索引已存在')

# 方式二
print(es.indices.create(index="test11",ignore=400))
print(es.indices.delete("test1"，ignore=[400, 404]))

插入数据时自动创建索引

# 创建索引，并向索引内插入数据，显示指定id，若id已存在则会更新数据，同时更新_version
print(es.index(index="test12", body={"title": "forver", "vote": 400}, id=1))
# 向索引内插入数据，未指定id，插入成功后会返回Id
print(es.index(index="test12", body={"title": "loving", "vote": 500}))

# 输出
/home/xadocker/PycharmProjects/untitled2/venv/bin/python /home/xadocker/PycharmProjects/untitled2/es_test.py
True
{'_index': 'test12', '_type': '_doc', '_id': '1', '_version': 2, 'result': 'updated', '_shards': {'total': 2, 'successful': 2, 'failed': 0}, '_seq_no': 2, '_primary_term': 1}
{'_index': 'test12', '_type': '_doc', '_id': 'U5eMJIUBrw8JBHMMY8mn', '_version': 1, 'result': 'created', '_shards': {'total': 2, 'successful': 2, 'failed': 0}, '_seq_no': 3, '_primary_term': 1}

若不想更新已存在id的数据，则可以使用creat方法创建

方法二: create
需要我们指定 id 字段来唯一标识该条数据
若该id存在,则报错
 
from elasticsearch.exceptions import ConflictError
try:
    es.create(index="test12",id=1,body={"title":"loving","vote":500})
except ConflictError as e:
    print('id存在')

删除数据

# 忽略待删除的数据不存在时
print(es.delete(index="test12",id=1,ignore=[404]))

# 另外一种方式
from elasticsearch.exceptions import NotFoundError
try:
    es.delete(index='user_info', doc_type='_doc', id=2)
except NotFoundError as e:
    print('数据不存在')

查询数据

# 查询test1索引所有数据
print(es.search(index="test12"))
# 查询单条数据1
print(es.get(index="test12", id=1))
# 使用SDL语法查询
print(es.search(body={
    "query": {
        "match": {
            "title": "loving"
        }
    }
}))

# 输出
/home/xadocker/PycharmProjects/untitled2/venv/bin/python /home/xadocker/PycharmProjects/untitled2/es_test.py
True
{'took': 3, 'timed_out': False, '_shards': {'total': 1, 'successful': 1, 'skipped': 0, 'failed': 0}, 'hits': {'total': {'value': 4, 'relation': 'eq'}, 'max_score': 1.0, 'hits': [{'_index': 'test12', '_type': '_doc', '_id': 'ZJeLJIUBrw8JBHMMjMiB', '_score': 1.0, '_source': {'title': 'dffff', 'vote': 500}}, {'_index': 'test12', '_type': '_doc', '_id': 'U5eMJIUBrw8JBHMMY8mn', '_score': 1.0, '_source': {'title': 'loving', 'vote': 500}}, {'_index': 'test12', '_type': '_doc', '_id': '1', '_score': 1.0, '_source': {'title': 'forver', 'vote': 400}}, {'_index': 'test12', '_type': '_doc', '_id': '2JeNJIUBrw8JBHMM5Mo6', '_score': 1.0, '_source': {'title': 'loving', 'vote': 500}}]}}
{'_index': 'test12', '_type': '_doc', '_id': '1', '_version': 3, '_seq_no': 4, '_primary_term': 1, 'found': True, '_source': {'title': 'forver', 'vote': 400}}
{'took': 97, 'timed_out': False, '_shards': {'total': 137, 'successful': 137, 'skipped': 0, 'failed': 0}, 'hits': {'total': {'value': 2, 'relation': 'eq'}, 'max_score': 1.1469179, 'hits': [{'_index': 'test12', '_type': '_doc', '_id': 'U5eMJIUBrw8JBHMMY8mn', '_score': 1.1469179, '_source': {'title': 'loving', 'vote': 500}}, {'_index': 'test12', '_type': '_doc', '_id': '2JeNJIUBrw8JBHMM5Mo6', '_score': 1.1469179, '_source': {'title': 'loving', 'vote': 500}}]}}

查询所有

body = {
        'query':{
        'match_all':{}
         }
    }

res = es.search(index='test12',body=body)
print(res['hits']['hits'])

根据字段值查询

body = {
        "query":{
            "term":{
                "vote": 400
                }
            }
        }
res = es.search(index='test12',body=body)
print(res['hits']['hits'])

根据字段多个值查询

body = {
        "query":{
            "term":{
                "vote": [400, 500]
                }
            }
        }
res = es.search(index='test12',body=body)
print(res['hits']['hits'])

字段值包含词查询

print(es.index(index="test12", body={"title": "loving forver", "vote": 600}))
print(es.index(index="test12", body={"title": "loving forver never", "vote": 500}))
body = {
    "query": {
        "match": {
            "title": "loving"
        }
    }
}
res = es.search(index='test12', body=body)
print(res['hits']['hits'])

范围查询

body = {
    "query": {
        "range": {
            "vote": {
                "gte": 300,
                "lte": 600
            }
        }
    }
}
res = es.search(index='test12', body=body)
print(res['hits']['hits'])

update数据

# 更新一条数据
body = {
    "doc": {
        "title": "loving"
    }
}
print(es.update(index="test12", id=1, body=body))
print(es.search(index="test12"))

# 输出
/home/xadocker/PycharmProjects/untitled2/venv/bin/python /home/xadocker/PycharmProjects/untitled2/es_test.py
True
/home/xadocker/PycharmProjects/untitled2/es_test.py:42: ElasticsearchDeprecationWarning: [types removal] Specifying types in document update requests is deprecated, use the endpoint /{index}/_update/{id} instead.
  print(es.update(index="test12", id=1, body=body))
{'_index': 'test12', '_type': '_doc', '_id': '1', '_version': 6, 'result': 'updated', '_shards': {'total': 2, 'successful': 2, 'failed': 0}, '_seq_no': 8, '_primary_term': 1}
{'took': 2, 'timed_out': False, '_shards': {'total': 1, 'successful': 1, 'skipped': 0, 'failed': 0}, 'hits': {'total': {'value': 4, 'relation': 'eq'}, 'max_score': 1.0, 'hits': [{'_index': 'test12', '_type': '_doc', '_id': '1', '_score': 1.0, '_source': {'title': 'frede', 'vote': 400}}, {'_index': 'test12', '_type': '_doc', '_id': 'U5eMJIUBrw8JBHMMY8mn', '_score': 1.0, '_source': {'title': 'loving', 'vote': 500}}, {'_index': 'test12', '_type': '_doc', '_id': '2JeNJIUBrw8JBHMM5Mo6', '_score': 1.0, '_source': {'title': 'loving', 'vote': 500}}, {'_index': 'test12', '_type': '_doc', '_id': 'ZJeLJIUBrw8JBHMMjMiB', '_score': 1.0, '_source': {'title': 'dffff', 'vote': 500}}]}}

创建索引时设置mapping

body = {
    'mappings': {
        'dynamic': 'strict',
        'properties': {
            'id': {
                'type': 'text',
            },
            'message': {
                'type': 'text',
                'analyzer': 'ik_max_word',
                "search_analyzer": "ik_smart"
            },
            'queue_id': {
                'type': 'text',
            },
        }
    }
}

res = es.indices.create(index='my-test-index', body=body)
print(res)
messages = [
    {'message': '后端服务异常，异步接口超时', 'id': 1, 'queue_id': 1},
    {'message': '用户注册失败', 'id': 2, 'queue_id': 2},
    {'message': '用户IP频繁注册，已增加限制', 'id': 2, 'queue_id': 3},
]

for dic in info:
    es.index(index='my-index', doc_type="_doc", body=dic)

查询message字段分词结果中，包含关键词“失败”或“超时”

body = {
  "query": {
    "match": {
      "message": {
        "query": "失败 超时",
        "operator": "or"
      }
    }
  }
}
res = es.search(index="my-test-index",body=body)
print(res['hits']['hits'])

# 输出
[{'_index': 'my-test-index', '_type': '_doc', '_id': 'wBzSJIUBW51jGgQU_LOx', '_score': 1.1946433, '_source': {'message': '用户注册失败', 'id': 2, 'queue_id': 2}}, {'_index': 'my-test-index', '_type': '_doc', '_id': 'Q_vSJIUBWeJoItpr_Kmp', '_score': 0.9331132, '_source': {'message': '后端服务异常，异步接口超时', 'id': 1, 'queue_id': 1}}]