扫码关注官方微信

扫码下载APP
返回顶部
首页 > 资讯 > 后端开发 > Python >如何使用python生成大量数据写入es数据库并查询操作
  • 812

0
分享到

如何使用python生成大量数据写入es数据库并查询操作

2024-04-02 19:04:59 812人浏览 泡泡鱼

Python 官方文档：入门教程 => 点击学习

摘要

前言： 模拟学生成绩信息写入es数据库，包括姓名、性别、科目、成绩。 示例代码1：【一次性写入10000*1000条数据】  【本人亲测耗时5100秒】 from elas

前言：

模拟学生成绩信息写入es数据库，包括姓名、性别、科目、成绩。

示例代码1：【一次性写入10000*1000条数据】  【本人亲测耗时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 = ['语文', '数学', '英语', '生物', '地理']
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数据库
# 批量写入数据
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条数据】  【本人亲测耗时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 = ['语文', '数学', '英语', '生物', '地理']
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数据库
# 批量写入数据
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条数据】  【耗时过长】

from elasticsearch import Elasticsearch
from elasticsearch import helpers
import random
import time
 
es = Elasticsearch(hosts='http://127.0.0.1:9200')
names = ['刘一', '陈二', '张三', '李四', '王五', '赵六', '孙七', '周八', '吴九', '郑十']
sexs = ['男', '女']
subjects = ['语文', '数学', '英语', '生物', '地理']
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数据库
# 批量写入数据
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)

查询数据并计算各种方式的成绩总分。

示例代码4：【一次性获取所有的数据，在程序中分别计算所耗的时间】

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)
 
    # 统计查询出来的所有学生的所有课程的所有成绩的总成绩
    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)
 
    # 统计查询出来的每个学生的所有课程的所有成绩的总成绩
    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)
 
    # 统计查询出来的每个学生的每门课程的所有成绩的总成绩
    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中所统计成绩的方法，面对大量的数据是比较耗时的，要使用es中的聚合查询。计算数据中所有成绩之和。

示例代码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)
 
    # 统计查询出来的所有学生的所有课程的所有成绩的总成绩
    start1 = time.time()
    all_grade = 0
    for data in all_source:
        all_grade += int(data['grade'])
    print('200万数据所有学生总成绩之和：', all_grade)
    end1 = time.time()
    print("耗时：", end1 - start1)
 
    end = time.time()
    print('200万数据总耗时：', 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万数据总成绩之和:', sum)
    end_aggs = time.time()
    print('1000万数据总耗时：', end_aggs - start_aggs)

运行结果：

计算数据中每个同学的各科总成绩之和。 

示例代码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)
 
    # 统计查询出来的每个学生的所有课程的所有成绩的总成绩
    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万数据耗时：", end2 - start2)
 
    end = time.time()
    print('200万数据总耗时：', 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万数据:', dic)
    end_aggs = time.time()
    print('1000万数据总耗时：', end_aggs - start_aggs)

运行结果：

计算数据中每个同学的每科成绩之和。 

示例代码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)
 
    # 统计查询出来的每个学生的每门课程的所有成绩的总成绩
    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万数据:', all_name_all_subject_grade)
    end3 = time.time()
    print("耗时：", end3 - start3)
    end = time.time()
    print('200万数据总耗时：', 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万数据:', dic)
    end_aggs = time.time()
    print('1000万数据总耗时：', end_aggs - start_aggs)

运行结果：

 在上面查询计算示例代码中，当使用含有1000万数据的索引grade时，普通方法查询计算是比较耗时的，使用聚合查询能够大大节约大量时间。当面对9205万数据的索引grade2时，这时使用普通计算方法所消耗的时间太大了，在线上开发环境中是不可用的，所以必须使用聚合方法来计算。

示例代码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万数据:', dic)
    end_aggs = time.time()
    print('9205万数据总耗时：', end_aggs - start_aggs)

运行结果：

注意：写查询语句时建议使用kibana去写，然后复制查询语句到代码中，kibana会提示查询语句。

到此这篇关于如何使用python生成大量数据写入es数据库并查询操作的文章就介绍到这了,更多相关Python es 内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！

您可能感兴趣的文档:

--结束END--

本文标题: 如何使用python生成大量数据写入es数据库并查询操作

本文链接: https://lsjlt.com/news/120481.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com    QQ/279061341

猜你喜欢
  • 如何使用python生成大量数据写入es数据库并查询操作
    前言： 模拟学生成绩信息写入es数据库，包括姓名、性别、科目、成绩。 示例代码1：【一次性写入10000*1000条数据】  【本人亲测耗时5100秒】 from elas...
    99+
    2024-04-02
  • 使用python生成大量数据写入es数据库并查询操作(2)
    目录方案一方案二1.顺序插入5000000条数据2.批量插入5000000条数据3.批量插入50000000条数据前言 ： 上一篇文章：如何使用python生成大量数据写入es数据库...
    99+
    2024-04-02
  • python如何生成EXCEL、连接数据库、并将指定数据写入EXCEL
    小编给大家分享一下python如何生成EXCEL、连接数据库、并将指定数据写入EXCEL，希望大家阅读完这篇文章之后都有所收获，下面让我们一起去探讨吧！#!/usr/bin/p...
    99+
    2024-04-02
  • 如何查询数据库数据量的大小
    这篇文章主要介绍如何查询数据库数据量的大小，文中介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们一定要看完！查询一下数据库的数据量的大小，自己做了个脚本：select sum(zon...
    99+
    2024-04-02
  • python将Dataframe格式的数据写入opengauss数据库并查询
    目录一、将数据写入opengauss二、python条件查询opengauss数据库中文列名的数据一、将数据写入opengauss 前提准备： 成功opengauss数据库，并创建用...
    99+
    2024-04-02
  • 如何使用PHP查询数据库数值进行操作
    PHP是一种非常流行的服务器端编程语言，广泛用于Web开发。在Web开发中，PHP通常与数据库一起使用，以更好地管理和操作数据。在本文中，我们将介绍如何使用PHP查询数据库数值进行操作。首先，我们需要连接到数据库。PHP提供了许多库和扩展来...
    99+
    2023-05-14
    php 数据库
  • Javascript如何连接数据库查询并插入数据
    这篇文章主要介绍了Javascript如何连接数据库查询并插入数据，具有一定借鉴价值，感兴趣的朋友可以参考下，希望大家阅读完这篇文章之后大有收获，下面让小编带着大家一起了解一下。一、准备数据库下载MySQL ,借助数据库图形工具或者在cmd...
    99+
    2023-06-29
  • python怎么将Dataframe格式的数据写入opengauss数据库并查询
    本文小编为大家详细介绍“python怎么将Dataframe格式的数据写入opengauss数据库并查询”，内容详细，步骤清晰，细节处理妥当，希望这篇“python怎么将Dataframe格式的数据写入opengauss数据库并查询”文章能...
    99+
    2023-06-30
  • php如何操作数据库查询相同的数据
    这篇文章主要介绍“php如何操作数据库查询相同的数据”，在日常操作中，相信很多人在php如何操作数据库查询相同的数据问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”php如何操作数据库查询相同的数据”的疑惑有所...
    99+
    2023-07-05
  • 用python操作mysql数据库（之简单查询操作）
    1、mysql安装    此处省略一万字.......2、pip安装MySQLdb模块sudo pip install mysql-python3、简单代码#!/us...
    99+
    2024-04-02
  • 怎么使用php写接口并查询数据库
    今天小编给大家分享一下怎么使用php写接口并查询数据库的相关知识点，内容详细，逻辑清晰，相信大部分人都还太了解这方面的知识，所以分享这篇文章给大家参考一下，希望大家阅读完这篇文章后有所收获，下面我们一起来了解一下吧。一、准备工作在开始之前，...
    99+
    2023-07-06
  • 如何查询mysql数据库中哪些表的数据量最大
    1.mysql的information_schema下有存储数据库基本信息的数据字典表，可以通过查询tables表来获得 所需要的表相关信息。  mysql> show ...
    99+
    2024-04-02
  • Teradata数据库如何处理大量数据的存储和查询
    Teradata数据库是一种用于存储和处理大量数据的关系型数据库管理系统。它采用了共享架构和并行处理技术，可以处理PB级别的数据量，...
    99+
    2024-04-02
  • Mysql如何实现查询数据库容量大小
    这篇文章主要为大家展示了Mysql如何实现查询数据库容量大小，内容简而易懂，希望大家可以学习一下，学习完之后肯定会有收获的，下面让小编带大家一起来看看吧。查询所有数据库的总大小方法如下：mysql> ...
    99+
    2024-04-02
  • 如何在VS Code连接MySql数据库并进行查询操作
    这期内容当中小编将会给大家带来有关如何在VS Code连接MySql数据库并进行查询操作，文章内容丰富且以专业的角度为大家分析和叙述，阅读完这篇文章希望大家可以有所收获。使用方法Step1. 安装 Visual Studio CodeVS ...
    99+
    2023-06-06
  • PostgreSQL数据库视图及子查询使用操作
    目录视图创建视图使用视图视图更新：删除视图：子查询关联子查询视图 表里面保存的是实际数据，视图里面保存的是SELECT语句(视图本身不存储数据)。 从视图中读取数据，此时视图...
    99+
    2024-04-02
  • 怎么使用PHP查询数据库数值进行操作
    这篇文章主要介绍“怎么使用PHP查询数据库数值进行操作”的相关知识，小编通过实际案例向大家展示操作过程，操作方法简单快捷，实用性强，希望这篇“怎么使用PHP查询数据库数值进行操作”文章能帮助大家解决问题。首先，我们需要连接到数据库。PHP提...
    99+
    2023-07-05
  • php如何操作数据库查询10条语句
    本篇内容介绍了“php如何操作数据库查询10条语句”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！查询单个字段当需要查询某一张表的单个字段时，...
    99+
    2023-07-05
  • 学习python之编写简单简单连接数据库并执行查询操作
    python 连接数据库操作, 方法如下: 在本机的mysql 数据库中有一个名为yao的库，其中有一个名为user的表，表中的内容如图 下面，则是python连接数据库的方法，及查找出表中的内容，代码如...
    99+
    2022-06-04
    简单 连接数据库 操作
  • 如何使用 PHP 查询数据库并显示结果
    使用 php 查询数据库并显示结果的步骤：连接数据库；查询数据库；显示结果，遍历查询结果的行并输出特定列数据。 如何使用 PHP 查询数据库并显示结果 使用 PHP 查询数据库并显示结...
    99+
    2024-05-02
    php 数据库查询 mysql
软考高级职称资格查询
热门wiki
近期文章
推荐阅读
热门问答
热门标签
编程网，编程工程师的家园，是目前国内优秀的开源技术社区之一，形成了由开源软件库、代码分享、资讯、协作翻译、讨论区和博客等几大频道内容，为IT开发者提供了一个发现、使用、并交流开源技术的平台。

  • 官方手机版

  • 微信公众号

  • 商务合作