ES

它是一个 分布式的 全文搜索与数据分析引擎。

与关系型数据库对比：
一个ES集群可以包含多个索引（数据库），每个索引又包含了很多类型（表），类型中包含了很多文档（行），每个文档使用 JSON 格式存储数据，包含了很多字段（列）

基本操作：

连接:

安装

pip install elasticsearch
1

连接

from elasticsearch import Elasticsearch

# 实例化
es = Elasticsearch([{"host": "ip", "port": 9200}])

1
2
3
4
5
6

插入:

创建数据库

es.indices.create(index="index_name", ignore=400)
1

创建数据库，并忽略400错误（数据库重复时，会返回400）

插入数据

body = {
	"name": "李李小明",
	"age": 7,
	"sex": "male"
}

es.index(index="es_test", doc_type='_doc', body=body)
1
2
3
4
5
6
7

插入多条

doc = [
	{"index": {'_index': "es_test", "_type": '_doc', '_id': 1}},
	{"name": "王王大志", "age": 8, "sex": "male"},
	{"index": {'_index': "es_test", "_type": '_doc', '_id': 2}},
	{"name": "刘刘大彪", "age": 8, "sex": "male"},
	{"index": {'_index': "es_test", "_type": '_doc', '_id': 3}},
	{"name": "赵赵灵儿", "age": 7, "sex": "male"},
]

es.bulk(index='es_test', doc_type='_doc', body=doc)
1
2
3
4
5
6
7
8
9
10

查询:

基本命令

# 返回字段
filter_path = [
	'hits.hits._source.ziduan1'
]

# 查询内容 过滤条件	
body = {
	"from"： 0,
	"size": 10
}

		
es.search(index='es_python'， filter_path=filter_path, body=body)
1
2
3
4
5
6
7
8
9
10
11
12
13
14

from， size 类似于 limit 0, 10

各种查询方式：
body 用法

模糊查询：match

body = {
	"query": {
		"match": {
			"name": "王"
		}
	},
	"size": 10
}
1
2
3
4
5
6
7
8

模糊匹配 字段 name 中有王的 文档。

精确查询： term

body = {
	"query": {
		"term": {
			"name.keyword": "刘刘大彪"
		}
	},
	"size": 10
}
1
2
3
4
5
6
7
8

精确多值查询：terms

body = {
	"query": {
		"terms": {
			"name.keyword": ["刘刘大彪", "赵赵灵儿" ]
		}
	},
	"size": 10
}
1
2
3
4
5
6
7
8

注意是或的关系

模糊多字段查询：multi_match

body = {
	"query": {
		"multi_match": {
			"query": "大"
			"fields":["name", "info"] # 指定字段
		}
	},
	"size": 10
}
1
2
3
4
5
6
7
8
9

前缀查询：prefix

body = {
	"query": {
		"prefix": {
			"name.keyword": "王"
		}
	}，
	size: 20
}
1
2
3
4
5
6
7
8

其他类似
通配符查询：wildcard
正则查询：regexp

布尔关系： bool

• must 相当于and
• should 相当于 or
• *_not 取反，
• 支持嵌套

body = {
	"query": {
		"bool": {
			"must": [
			
				{"term": xxxxxxxx},
				{"prefix": xxxx},
				
				{
					"bool": {
						"should": [
							{"term": xxxxxx},
							{"term": xxxxxx}
						]
					}
				}
			
			]
		}
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

连接池

import os
import json
from datetime import datetime
from elasticsearch import Elasticsearch, RequestsHttpConnection
from elasticsearch import Transport
from elasticsearch.exceptions import NotFoundError

pool = Transport(
				 hosts=hosts, 
				 connection_class=RequestsHttpConnection
		).connection_pool

conn = pool.get_connection()
1
2
3
4
5
6
7
8
9
10
11
12
13

一个工具类： 写的不错，直接就复制过来了

小建议: 对于查询直接写成对应函数，并提供连接池管理。

import os
import json
from datetime import datetime
from elasticsearch import Elasticsearch, RequestsHttpConnection
from elasticsearch import Transport
from elasticsearch.exceptions import NotFoundError
 
 
class ES(object):
 
    _index = ""
    _type = ""
    
    def __init__(self, hosts):
        # 基于requests实例化es连接池
        self.conn_pool = Transport(hosts=hosts, connection_class=RequestsHttpConnection).connection_pool
 
    def get_conn(self):
        """
        从连接池获取一个连接
        """
        conn = self.conn_pool.get_connection()
        return conn
 
    def request(self, method, url, headers=None, params=None, body=None):
        """
        想es服务器发送一个求情
        @method     请求方式
        @url        请求的绝对url  不包括域名
        @headers    请求头信息
        @params     请求的参数：dict
        @body       请求体：json对象(headers默认Content-Type为application/json)
        # return    返回体：python内置数据结构
        """
        conn = self.get_conn()
        try:
            status, headers, body = conn.perform_request(method, url, headers=headers, params=params, body=body)
        except NotFoundError as e:
            return None
        if method == "HEAD":
            return status
        return json.loads(body)
 
    def search(self, query=None, method="GET"):
        url = "/%s/%s/_search" % (self._index, self._type)
        if method == "GET":
            data = self.get(url, params=query)
        elif method == "POST":
            data = self.post(url, body=query)
        else:
            return None
        return data
 
    def get(self, url, params=None, method="GET"):
        """
        使用get请求访问es服务器
        """
        data = self.request(method, url, params=params)
        return data
 
    def put(self, url, body=None, method="PUT"):
        """
        使用put请求访问es
        """
        data = self.request(method, url, body=body)
        return data
 
    def post(self, url, body=None, method="POST"):
        """使用post请求访问服务器"""
        data = self.request(method, url, body=body)
        return data
 
    def head(self, url, *args, **kwargs):
        status = self.request("HEAD", url, *args, **kwargs)
        return status
 
    def delete(self, url, *args, **kwargs):
        ret = self.request("DELETE", url, *args, **kwargs)
        return ret
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
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

进阶

es 做到数十亿数据查询毫秒级响应

往es中写文件 最终是保存在硬盘上， 查询的时候 是从硬盘上读数据。然而：

操作系统会把从硬盘上读出的数据放到filesystem cache 中， 下次读取的时候直接从缓存中取数据， 从内存中取数据明显要快于从硬盘读取，性能会得到极大的提升。

所以在设计（调优）时：

首先：硬件方面
一般至少确保60% - 80% 的数据能够缓存到 系统缓存中， 这样整个系统性能才能有保证。 并且尽量保证运行es 的机器上 不运行其他程序，避免cpu 与内存的争用。

其次： 设计方面
冷热分离，在某些情况下， 数据确实很多，不能将大部分数据放大系统缓存。根据业务场景，一般热数据占比是比较少的， 也就是大量数据中，经常访问的数据量不多，但这批数据访问频次很高。所以只要保证将这批热数据放到系统缓存中，就能保证系统中大部分用户的使用体验。
因此，可以将es中冷热数据分离，给他们分配单独机器，并适当调大内存，保证热数据访问效率。 对于剩下的大批冷数据可以 放到另外机器中，虽然查询会多花费一些时间，但这些是一小批用户。 对于系统整体来说，在设备条件有限的情况下 能最大程度使用硬件。

再者：编码方面
文档设计方面，在机器性能有限的情况下， 应避免存放与查询无关的内容，尽可能减小单个文档大小，增大缓存文档数量。
例如: 商品搜索中，只存放 id title keyword， 每次查询，根据条件查询出商品id ，然后根据id 从其他数据库中查询出详细信息。

另外为提升用户体验，可以进行数据预热，始终保证大部分热数据存在于系统缓存中。

分页性能优化，使用scroll , 一页一页 返回， 并对分页深度进行限制。

另外，集群条件下，提高系统查询效率，可以多添加一些副本，分担系统压力。

---

问题： 分布式查询数据，较深的分页 为什么性能很差

问题： 怎么找出系统热数据

---

参考文章： xx 官方文档