網(wǎng)站首頁 編程語言 正文
前言：
模擬學生成績信息寫入es數(shù)據(jù)庫，包括姓名、性別、科目、成績。
示例代碼1：【一次性寫入10000*1000條數(shù)據(jù)】? 【本人親測耗時5100秒】
from elasticsearch import Elasticsearch
from elasticsearch import helpers
import random
import time
es = Elasticsearch(hosts='http://127.0.0.1:9200')
# print(es)
names = ['劉一', '陳二', '張三', '李四', '王五', '趙六', '孫七', '周八', '吳九', '鄭十']
sexs = ['男', '女']
subjects = ['語文', '數(shù)學', '英語', '生物', '地理']
grades = [85, 77, 96, 74, 85, 69, 84, 59, 67, 69, 86, 96, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86]
datas = []
start = time.time()
# 開始批量寫入es數(shù)據(jù)庫
# 批量寫入數(shù)據(jù)
for j in range(1000):
print(j)
action = [
{
"_index": "grade",
"_type": "doc",
"_id": i,
"_source": {
"id": i,
"name": random.choice(names),
"sex": random.choice(sexs),
"subject": random.choice(subjects),
"grade": random.choice(grades)
}
} for i in range(10000 * j, 10000 * j + 10000)
]
helpers.bulk(es, action)
end = time.time()
print('花費時間：', end - start)elasticsearch-head中顯示：
示例代碼2：【一次性寫入10000*5000條數(shù)據(jù)】? 【本人親測耗時23000秒】
from elasticsearch import Elasticsearch
from elasticsearch import helpers
import random
import time
es = Elasticsearch(hosts='http://127.0.0.1:9200')
# print(es)
names = ['劉一', '陳二', '張三', '李四', '王五', '趙六', '孫七', '周八', '吳九', '鄭十']
sexs = ['男', '女']
subjects = ['語文', '數(shù)學', '英語', '生物', '地理']
grades = [85, 77, 96, 74, 85, 69, 84, 59, 67, 69, 86, 96, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86]
datas = []
start = time.time()
# 開始批量寫入es數(shù)據(jù)庫
# 批量寫入數(shù)據(jù)
for j in range(5000):
print(j)
action = [
{
"_index": "grade3",
"_type": "doc",
"_id": i,
"_source": {
"id": i,
"name": random.choice(names),
"sex": random.choice(sexs),
"subject": random.choice(subjects),
"grade": random.choice(grades)
}
} for i in range(10000 * j, 10000 * j + 10000)
]
helpers.bulk(es, action)
end = time.time()
print('花費時間：', end - start)
示例代碼3：【一次性寫入10000*9205條數(shù)據(jù)】? 【耗時過長】
from elasticsearch import Elasticsearch
from elasticsearch import helpers
import random
import time
es = Elasticsearch(hosts='http://127.0.0.1:9200')
names = ['劉一', '陳二', '張三', '李四', '王五', '趙六', '孫七', '周八', '吳九', '鄭十']
sexs = ['男', '女']
subjects = ['語文', '數(shù)學', '英語', '生物', '地理']
grades = [85, 77, 96, 74, 85, 69, 84, 59, 67, 69, 86, 96, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86]
datas = []
start = time.time()
# 開始批量寫入es數(shù)據(jù)庫
# 批量寫入數(shù)據(jù)
for j in range(9205):
print(j)
action = [
{
"_index": "grade2",
"_type": "doc",
"_id": i,
"_source": {
"id": i,
"name": random.choice(names),
"sex": random.choice(sexs),
"subject": random.choice(subjects),
"grade": random.choice(grades)
}
} for i in range(10000*j, 10000*j+10000)
]
helpers.bulk(es, action)
end = time.time()
print('花費時間：', end - start)
查詢數(shù)據(jù)并計算各種方式的成績總分。
示例代碼4：【一次性獲取所有的數(shù)據(jù)，在程序中分別計算所耗的時間】
from elasticsearch import Elasticsearch
import time
def search_data(es, size=10):
query = {
"query": {
"match_all": {}
}
}
res = es.search(index='grade', body=query, size=size)
# print(res)
return res
if __name__ == '__main__':
start = time.time()
es = Elasticsearch(hosts='http://192.168.1.1:9200')
# print(es)
size = 10000
res = search_data(es, size)
# print(type(res))
# total = res['hits']['total']['value']
# print(total)
all_source = []
for i in range(size):
source = res['hits']['hits'][i]['_source']
all_source.append(source)
# print(source)
# 統(tǒng)計查詢出來的所有學生的所有課程的所有成績的總成績
start1 = time.time()
all_grade = 0
for data in all_source:
all_grade += int(data['grade'])
print('所有學生總成績之和：', all_grade)
end1 = time.time()
print("耗時：", end1 - start1)
# 統(tǒng)計查詢出來的每個學生的所有課程的所有成績的總成績
start2 = time.time()
names1 = []
all_name_grade = {}
for data in all_source:
if data['name'] in names1:
all_name_grade[data['name']] += data['grade']
else:
names1.append(data['name'])
all_name_grade[data['name']] = data['grade']
print(all_name_grade)
end2 = time.time()
print("耗時：", end2 - start2)
# 統(tǒng)計查詢出來的每個學生的每門課程的所有成績的總成績
start3 = time.time()
names2 = []
subjects = []
all_name_all_subject_grade = {}
for data in all_source:
if data['name'] in names2:
if all_name_all_subject_grade[data['name']].get(data['subject']):
all_name_all_subject_grade[data['name']][data['subject']] += data['grade']
else:
all_name_all_subject_grade[data['name']][data['subject']] = data['grade']
else:
names2.append(data['name'])
all_name_all_subject_grade[data['name']] = {}
all_name_all_subject_grade[data['name']][data['subject']] = data['grade']
print(all_name_all_subject_grade)
end3 = time.time()
print("耗時：", end3 - start3)
end = time.time()
print('總耗時：', end - start)運行結果：
在示例代碼4中當把size由10000改為 2000000時，運行效果如下所示：
在項目中一般不用上述代碼4中所統(tǒng)計成績的方法，面對大量的數(shù)據(jù)是比較耗時的，要使用es中的聚合查詢。計算數(shù)據(jù)中所有成績之和。
示例代碼5：【使用普通計算方法和聚類方法做對比驗證】
from elasticsearch import Elasticsearch
import time
def search_data(es, size=10):
query = {
"query": {
"match_all": {}
}
}
res = es.search(index='grade', body=query, size=size)
# print(res)
return res
def search_data2(es, size=10):
query = {
"aggs": {
"all_grade": {
"terms": {
"field": "grade",
"size": 1000
}
}
}
}
res = es.search(index='grade', body=query, size=size)
# print(res)
return res
if __name__ == '__main__':
start = time.time()
es = Elasticsearch(hosts='http://127.0.0.1:9200')
size = 2000000
res = search_data(es, size)
all_source = []
for i in range(size):
source = res['hits']['hits'][i]['_source']
all_source.append(source)
# print(source)
# 統(tǒng)計查詢出來的所有學生的所有課程的所有成績的總成績
start1 = time.time()
all_grade = 0
for data in all_source:
all_grade += int(data['grade'])
print('200萬數(shù)據(jù)所有學生總成績之和：', all_grade)
end1 = time.time()
print("耗時：", end1 - start1)
end = time.time()
print('200萬數(shù)據(jù)總耗時：', end - start)
# 聚合操作
start_aggs = time.time()
es = Elasticsearch(hosts='http://127.0.0.1:9200')
# size = 2000000
size = 0
res = search_data2(es, size)
# print(res)
aggs = res['aggregations']['all_grade']['buckets']
print(aggs)
sum = 0
for agg in aggs:
sum += (agg['key'] * agg['doc_count'])
print('1000萬數(shù)據(jù)總成績之和:', sum)
end_aggs = time.time()
print('1000萬數(shù)據(jù)總耗時：', end_aggs - start_aggs)運行結果：
計算數(shù)據(jù)中每個同學的各科總成績之和。?
示例代碼6：? 【子聚合】【先分組，再計算】
from elasticsearch import Elasticsearch
import time
def search_data(es, size=10):
query = {
"query": {
"match_all": {}
}
}
res = es.search(index='grade', body=query, size=size)
# print(res)
return res
def search_data2(es):
query = {
"size": 0,
"aggs": {
"all_names": {
"terms": {
"field": "name.keyword",
"size": 10
},
"aggs": {
"total_grade": {
"sum": {
"field": "grade"
}
}
}
}
}
}
res = es.search(index='grade', body=query)
# print(res)
return res
if __name__ == '__main__':
start = time.time()
es = Elasticsearch(hosts='http://127.0.0.1:9200')
size = 2000000
res = search_data(es, size)
all_source = []
for i in range(size):
source = res['hits']['hits'][i]['_source']
all_source.append(source)
# print(source)
# 統(tǒng)計查詢出來的每個學生的所有課程的所有成績的總成績
start2 = time.time()
names1 = []
all_name_grade = {}
for data in all_source:
if data['name'] in names1:
all_name_grade[data['name']] += data['grade']
else:
names1.append(data['name'])
all_name_grade[data['name']] = data['grade']
print(all_name_grade)
end2 = time.time()
print("200萬數(shù)據(jù)耗時：", end2 - start2)
end = time.time()
print('200萬數(shù)據(jù)總耗時：', end - start)
# 聚合操作
start_aggs = time.time()
es = Elasticsearch(hosts='http://127.0.0.1:9200')
res = search_data2(es)
# print(res)
aggs = res['aggregations']['all_names']['buckets']
# print(aggs)
dic = {}
for agg in aggs:
dic[agg['key']] = agg['total_grade']['value']
print('1000萬數(shù)據(jù):', dic)
end_aggs = time.time()
print('1000萬數(shù)據(jù)總耗時：', end_aggs - start_aggs)運行結果：
計算數(shù)據(jù)中每個同學的每科成績之和。?
示例代碼7：
from elasticsearch import Elasticsearch
import time
def search_data(es, size=10):
query = {
"query": {
"match_all": {}
}
}
res = es.search(index='grade', body=query, size=size)
# print(res)
return res
def search_data2(es):
query = {
"size": 0,
"aggs": {
"all_names": {
"terms": {
"field": "name.keyword",
"size": 10
},
"aggs": {
"all_subjects": {
"terms": {
"field": "subject.keyword",
"size": 5
},
"aggs": {
"total_grade": {
"sum": {
"field": "grade"
}
}
}
}
}
}
}
}
res = es.search(index='grade', body=query)
# print(res)
return res
if __name__ == '__main__':
start = time.time()
es = Elasticsearch(hosts='http://127.0.0.1:9200')
size = 2000000
res = search_data(es, size)
all_source = []
for i in range(size):
source = res['hits']['hits'][i]['_source']
all_source.append(source)
# print(source)
# 統(tǒng)計查詢出來的每個學生的每門課程的所有成績的總成績
start3 = time.time()
names2 = []
subjects = []
all_name_all_subject_grade = {}
for data in all_source:
if data['name'] in names2:
if all_name_all_subject_grade[data['name']].get(data['subject']):
all_name_all_subject_grade[data['name']][data['subject']] += data['grade']
else:
all_name_all_subject_grade[data['name']][data['subject']] = data['grade']
else:
names2.append(data['name'])
all_name_all_subject_grade[data['name']] = {}
all_name_all_subject_grade[data['name']][data['subject']] = data['grade']
print('200萬數(shù)據(jù):', all_name_all_subject_grade)
end3 = time.time()
print("耗時：", end3 - start3)
end = time.time()
print('200萬數(shù)據(jù)總耗時：', end - start)
# 聚合操作
start_aggs = time.time()
es = Elasticsearch(hosts='http://127.0.0.1:9200')
res = search_data2(es)
# print(res)
aggs = res['aggregations']['all_names']['buckets']
# print(aggs)
dic = {}
for agg in aggs:
dic[agg['key']] = {}
for sub in agg['all_subjects']['buckets']:
dic[agg['key']][sub['key']] = sub['total_grade']['value']
print('1000萬數(shù)據(jù):', dic)
end_aggs = time.time()
print('1000萬數(shù)據(jù)總耗時：', end_aggs - start_aggs)運行結果：
?在上面查詢計算示例代碼中，當使用含有1000萬數(shù)據(jù)的索引grade時，普通方法查詢計算是比較耗時的，使用聚合查詢能夠大大節(jié)約大量時間。當面對9205萬數(shù)據(jù)的索引grade2時，這時使用普通計算方法所消耗的時間太大了，在線上開發(fā)環(huán)境中是不可用的，所以必須使用聚合方法來計算。
示例代碼8：
from elasticsearch import Elasticsearch
import time
def search_data(es):
query = {
"size": 0,
"aggs": {
"all_names": {
"terms": {
"field": "name.keyword",
"size": 10
},
"aggs": {
"all_subjects": {
"terms": {
"field": "subject.keyword",
"size": 5
},
"aggs": {
"total_grade": {
"sum": {
"field": "grade"
}
}
}
}
}
}
}
}
res = es.search(index='grade2', body=query)
# print(res)
return res
if __name__ == '__main__':
# 聚合操作
start_aggs = time.time()
es = Elasticsearch(hosts='http://127.0.0.1:9200')
res = search_data(es)
# print(res)
aggs = res['aggregations']['all_names']['buckets']
# print(aggs)
dic = {}
for agg in aggs:
dic[agg['key']] = {}
for sub in agg['all_subjects']['buckets']:
dic[agg['key']][sub['key']] = sub['total_grade']['value']
print('9205萬數(shù)據(jù):', dic)
end_aggs = time.time()
print('9205萬數(shù)據(jù)總耗時：', end_aggs - start_aggs)運行結果：
注意：寫查詢語句時建議使用kibana去寫，然后復制查詢語句到代碼中，kibana會提示查詢語句。
原文鏈接：https://blog.csdn.net/weixin_44799217/article/details/126794236
相關推薦
- 2022-04-23 通過自定義指令實現(xiàn) element-ui的tooltip組件 文本長度超出顯示不超出不顯示
- 2022-09-20 Flink實踐Savepoint使用示例詳解_服務器其它
- 2022-09-21 python裝飾器底層原理詳解_python
- 2022-07-22 idea 編譯項目后target包沒有resources文件
- 2022-06-01 python畫立方體--魔方_python
- 2022-07-20 C語言雙向鏈表的原理與使用操作_C 語言
- 2022-06-19 C++簡明講解類型轉換的使用與作用_C 語言
- 2022-07-21 C語言運算符深入探究優(yōu)先級與結合性及種類_C 語言
- 最近更新
-
- window11 系統(tǒng)安裝 yarn
- 超詳細win安裝深度學習環(huán)境2025年最新版（
- Linux 中運行的top命令 怎么退出？
- MySQL 中decimal 的用法？ 存儲小
- get 、set 、toString 方法的使
- @Resource和 @Autowired注解
- Java基礎操作-- 運算符，流程控制 Flo
- 1. Int 和Integer 的區(qū)別，Jav
- spring @retryable不生效的一種
- Spring Security之認證信息的處理
- Spring Security之認證過濾器
- Spring Security概述快速入門
- Spring Security之配置體系
- 【SpringBoot】SpringCache
- Spring Security之基于方法配置權
- redisson分布式鎖中waittime的設
- maven:解決release錯誤：Artif
- restTemplate使用總結
- Spring Security之安全異常處理
- MybatisPlus優(yōu)雅實現(xiàn)加密？
- Spring ioc容器與Bean的生命周期。
- 【探索SpringCloud】服務發(fā)現(xiàn)-Nac
- Spring Security之基于HttpR
- Redis 底層數(shù)據(jù)結構-簡單動態(tài)字符串（SD
- arthas操作spring被代理目標對象命令
- Spring中的單例模式應用詳解
- 聊聊消息隊列，發(fā)送消息的4種方式
- bootspring第三方資源配置管理
- GIT同步修改后的遠程分支
提供CDN加速