欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

如何使用python生成大量數(shù)據(jù)寫入es數(shù)據(jù)庫并查詢操作

 更新時間:2022年09月15日 16:19:11   作者:IT之一小佬  
這篇文章主要介紹了如何使用python生成大量數(shù)據(jù)寫入es數(shù)據(jù)庫并查詢操作,文章圍繞主題展開詳細的內(nèi)容介紹,具有一定的參考價值,需要的小伙伴可以參考一下

前言:

模擬學(xué)生成績信息寫入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ù)學(xué)', '英語', '生物', '地理']
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ù)學(xué)', '英語', '生物', '地理']
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ù)學(xué)', '英語', '生物', '地理']
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)計查詢出來的所有學(xué)生的所有課程的所有成績的總成績
    start1 = time.time()
    all_grade = 0
    for data in all_source:
        all_grade += int(data['grade'])
    print('所有學(xué)生總成績之和:', all_grade)
    end1 = time.time()
    print("耗時:", end1 - start1)
 
    # 統(tǒng)計查詢出來的每個學(xué)生的所有課程的所有成績的總成績
    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)計查詢出來的每個學(xué)生的每門課程的所有成績的總成績
    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)

運行結(jié)果:

在示例代碼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)計查詢出來的所有學(xué)生的所有課程的所有成績的總成績
    start1 = time.time()
    all_grade = 0
    for data in all_source:
        all_grade += int(data['grade'])
    print('200萬數(shù)據(jù)所有學(xué)生總成績之和:', 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)

運行結(jié)果:

計算數(shù)據(jù)中每個同學(xué)的各科總成績之和。 

示例代碼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)計查詢出來的每個學(xué)生的所有課程的所有成績的總成績
    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)

運行結(jié)果:

計算數(shù)據(jù)中每個同學(xué)的每科成績之和。 

示例代碼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)計查詢出來的每個學(xué)生的每門課程的所有成績的總成績
    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)

運行結(jié)果:

 在上面查詢計算示例代碼中,當使用含有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)

運行結(jié)果:

注意:寫查詢語句時建議使用kibana去寫,然后復(fù)制查詢語句到代碼中,kibana會提示查詢語句。

到此這篇關(guān)于如何使用python生成大量數(shù)據(jù)寫入es數(shù)據(jù)庫并查詢操作的文章就介紹到這了,更多相關(guān)python es 內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!

相關(guān)文章

  • Python可視化分析全球火山分布

    Python可視化分析全球火山分布

    也就在前幾天,南太平洋島國湯加發(fā)生火山噴發(fā)。所以今天小編將為大家介紹如何用Python當中的folium模塊以及其他的可視化庫來對全球的火山情況做一個分析。需要的可以參考一下
    2022-01-01
  • 在python list中篩選包含字符的字段方式

    在python list中篩選包含字符的字段方式

    這篇文章主要介紹了在python list中篩選包含字符的字段方式,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教
    2022-11-11
  • 快速排序的算法思想及Python版快速排序的實現(xiàn)示例

    快速排序的算法思想及Python版快速排序的實現(xiàn)示例

    快速排序算法來源于分治法的思想策略,這里我們將來為大家簡單解析一下快速排序的算法思想及Python版快速排序的實現(xiàn)示例:
    2016-07-07
  • 詳細解讀Python中的__init__()方法

    詳細解讀Python中的__init__()方法

    這篇文章主要介紹了Python中的__init__()方法,__init__()方法是Python學(xué)習當中重要的基礎(chǔ)知識,需要的朋友可以參考下
    2015-05-05
  • python批量處理文件或文件夾

    python批量處理文件或文件夾

    這篇文章主要為大家詳細介紹了python批量處理文件或文件夾,文中示例代碼介紹的非常詳細,具有一定的參考價值,感興趣的小伙伴們可以參考一下
    2019-08-08
  • Python中包的用法及安裝

    Python中包的用法及安裝

    在本篇文章里小編給大家整理的是關(guān)于Python中一些包的基本用處和安裝方法,需要的朋友們可以學(xué)習參考下。
    2020-02-02
  • 深入淺析Python中l(wèi)ist的復(fù)制及深拷貝與淺拷貝

    深入淺析Python中l(wèi)ist的復(fù)制及深拷貝與淺拷貝

    這篇文章主要介紹了Python中l(wèi)ist的復(fù)制及深拷貝與淺拷貝及區(qū)別解析 ,需要的朋友可以參考下
    2018-09-09
  • python計算質(zhì)數(shù)的6種方法

    python計算質(zhì)數(shù)的6種方法

    本文主要介紹了python計算質(zhì)數(shù)的6種方法,文中通過示例代碼介紹的非常詳細,對大家的學(xué)習或者工作具有一定的參考學(xué)習價值,需要的朋友們下面隨著小編來一起學(xué)習學(xué)習吧
    2023-05-05
  • Python中關(guān)于浮點數(shù)的冷知識

    Python中關(guān)于浮點數(shù)的冷知識

    這篇文章主要給大家介紹了Python中關(guān)于浮點數(shù)的冷知識,文中通過示例代碼介紹的非常詳細,對大家學(xué)習或者使用Python具有一定的參考學(xué)習價值,需要的朋友們下面來一起學(xué)習學(xué)習吧
    2019-09-09
  • Python3爬蟲之自動查詢天氣并實現(xiàn)語音播報

    Python3爬蟲之自動查詢天氣并實現(xiàn)語音播報

    這篇文章主要介紹了Python3爬蟲之自動查詢天氣并實現(xiàn)語音播報,文中通過示例代碼介紹的非常詳細,對大家的學(xué)習或者工作具有一定的參考學(xué)習價值,需要的朋友們下面隨著小編來一起學(xué)習學(xué)習吧
    2019-02-02

最新評論