--- title: 【好文推荐】openGauss psycopg2 批量操作性能测试-官方技术文章-鲲鹏社区 description: 本文主要向大家介绍如何对openGauss psycopg2批量操作性能进行测试。 keywords: Gauss,好文推荐,批量操作性能测试,官方技术文章,鲲鹏社区,测试版本,测试环境,测试接口 url: https://www.hikunpeng.com/developer/techArticles/20230921-2?envFlag=1 section: (其他) --- # 【好文推荐】openGauss psycopg2 批量操作性能测试-官方技术文章-鲲鹏社区 URL: https://www.hikunpeng.com/developer/techArticles/20230921-2?envFlag=1 描述: 本文主要向大家介绍如何对openGauss psycopg2批量操作性能进行测试。 关键词: Gauss,好文推荐,批量操作性能测试,官方技术文章,鲲鹏社区,测试版本,测试环境,测试接口 官方技术文章 [了解详情](https://www.hikunpeng.com/zh/developer/techArticles) 【好文推荐】openGauss psycopg2 批量操作性能测试 【好文推荐】openGauss psycopg2 批量操作性能测试 实战经验 发表于 2023/09/08 0 ## 测试版本 本测试基于 openGauss 版本的 psycopg2 驱动。 ``` import psycopg2 as pg >>> pg.__libpq_version__ 90204 >>> pg.__version__ '2.8.6 (dt dec pq3 ext)' ``` ## 测试环境 | 组件 | 说明 | | --- | ---| | 客户端 | Rocky Linux 8 虚拟机 | | 数据库 | openGauss 3.0.3 in docker | | 网络 | 本地回路网卡 | | Python | 3.6.8 | ## 测试接口 | 接口名 | 说明 | 备注 | | --- | --- | ---| | cursor.executemany(query, vars_list) | 执行一个数据库操作，vars_list 列表中的所有参数会逐个被应用到query 中，每组参数都会单独封包给服务端。 | 该函数主要用于更新数据库的命令，查询返回的任何结果集都将被丢弃。在其当前实现中，此方法并不比在循环中执行execute()快。 | | psycopg2.extras.execute_batch(cur, sql, argslist, page_size=100) | 批量执行一个数据库操作，执行的SQL和 executemany 相同，只是单个数据包时会一批SQL，数量由page_size决定。这样可以减少和服务端的通信次数 | execute_batch()也可以和预处理语句(PREPARE, EXECUTE, DEALLOCATE)一起使用。 | | extras.execute_batch + 预处理语句 | 使用PREPARE提交创建一个statement，然后通过 execute_batch 提交 | - | | psycopg2.extras.execute_values(cur, sql, argslist, template=None, page_size=100, fetch=False) | 将参数和SQL封装为一条SQL执行，单条SQL中参数的个数由 page_size 决定。 | - | ## 性能对比 ### 1. INSERT 测试数据 | rows | executemany | execute_batch | prepare+execute_batch | execute_values | | --- | --- | --- | --- | ---| | 10,000 | 9.782 | 0.707 | 0.501 | 0.266 | | 50,000 | 52.979 | 3.123 | 2.637 | 1.226 | | 100,000 | 111.504 | 6.831 | 4.557 | 2.125 | INSERT耗时对比图 INSERT 去除 executemany 对比 ### 2. UPDATE 测试数据 | rows | executemany | execute_batch | prepare+execute_batch | execute_values | | --- | --- | --- | --- | ---| | 10,000 | 5.015 | 0.617 | 0.425 | 0.356 | | 50,000 | 24.639 | 3.467 | 1.905 | 5.237 | | 100,000 | 52.095 | 6.927 | 3.473 | 21.102 | UPDATE 耗时对比图 ### 3. DELETE 测试数据 说明：100000 条数据组耗时太久不做展示 | rows | executemany | execute_batch | prepare+execute_batch | execute_values | | --- | --- | --- | --- | ---| | 10,000 | 15.02 | 8.699 | 0.277 | 6.204 | | 50,000 | 248.154 | 227.958 | 1.455 | 142.732 | DELETE 耗时对比图 ## 性能分析 从耗时对比来看，插入、更新、删除在不同的数据量情况下性能是不同的，用户应该根据自己的业务场景来选择使用哪一种操作接口。 插入性能从低到高依次为： ``` executemany < execute_batch < prepare+execute_batch < execute_values ``` 更新性能从低到高依次为： ``` executemany < execute_values < execute_batch < prepare+execute_batch ``` 删除性能从低到高依次为： ``` executemany < execute_batch < execute_values < prepare+execute_batch ``` 性能的高低主要是由于在向服务端数据包时的方式不同导致，下面以插入的SQL为例，通过 wireshark 进行抓包可以看出 psycopg2 在通信过程中不同批处理接口的封包情况。 ### 1. executemany executemany 提交SQL的时候是逐个应用给的参数，每个SQL都单独给服务端 ### 2. execute_batch execute_batch 接口区别于 executemany 的是，在给后端的单个请求包里的数据会一次性提交一批的SQL，这样可以减少和服务器之间通信的往返次数 ### 3. prepare+execute_batch prepare 可以提前在数据库里面创建一个预备语句对象，在执行 prepare 语句的时候，指定的SQL已经经了解析、分析、重写，这样在后续执行 EXECUTE 时就避免了重复解析分析的工作，从而起到优化性能的作用。 ### 4. execute_values 前面的三个接口，不管是单个提交还是批量提交，最终都是一行数据一个SQL到服务端的，所以服务端需要逐个执行，而 execute_values 接口是会按照 page_size 分组参数后，每组参数一次性组成一个SQL进行提交。 ## 测试代码 执行方式：python test.py （1） ：支持 executemany, execute_batch, prepare, execute_values （2）： 支持 insert, update, delete ``` # coding: utf-8 # Usage: python test.py import time import sys import psycopg2 as pg from psycopg2.extras import execute_batch, execute_values from contextlib import contextmanager if sys.argv[3] == "insert": args = [[str(i), i] for i in range(int(sys.argv[2]))] elif sys.argv[3] == "update": args = [[i, str(i)] for i in range(int(sys.argv[2]))] elif sys.argv[3] == "delete": args = [[i] for i in range(int(sys.argv[2]))] ''' - *dbname*: the database name - *database*: the database name (only as keyword argument) - *user*: user name used to authenticate - *password*: password used to authenticate - *host*: database host address (defaults to UNIX socket if not provided) - *port*: connection port number (defaults to 5432 if not provided) ''' conf = { 'dbname': "postgres", 'user': 'gaussdb', 'password': '', 'host': '', 'port': 26000, 'sslmode': 'disable' } @contextmanager def calc_time(s): start = time.time() yield end = time.time() print(f"{s} of '{sys.argv[3]}' cost: ", end - start) sql_map = { "insert": { 1: "INSERT INTO t_psycopg2_benchmark VALUES (%s, %s)", 2: "INSERT INTO t_psycopg2_benchmark VALUES ($1, $2)", 3: "INSERT INTO t_psycopg2_benchmark VALUES %s", }, "update": { 1: "UPDATE t_psycopg2_benchmark as t SET f_value = %s WHERE t.f_key = %s", 2: "UPDATE t_psycopg2_benchmark as t SET f_value = $1 WHERE t.f_key = $2", 3: "UPDATE t_psycopg2_benchmark as t SET f_value = data.v1 FROM (VALUES %s) AS data (id, v1) WHERE t.f_key = data.id", }, "delete": { 1: "DELETE FROM t_psycopg2_benchmark as t WHERE t.f_key=%s", 2: "DELETE FROM t_psycopg2_benchmark as t WHERE t.f_key=$1", 3: "DELETE FROM t_psycopg2_benchmark as t WHERE t.f_key IN (%s)", } } def insert_data(conn): print("* preparing data ...") args = [[str(i), i] for i in range(int(sys.argv[2]))] cursor = conn.cursor() sql = "insert into t_psycopg2_benchmark values %s" execute_values(cursor, sql, args) conn.commit() def main(): try: conn = pg.connect(**conf) print("* connect success") except Exception as e: print(f"connect failed: {e}") return cursor = conn.cursor() sql = "drop table if exists t_psycopg2_benchmark" cursor.execute(sql) sql = "create table t_psycopg2_benchmark (f_key text primary key, f_value numeric)" cursor.execute(sql) api = sys.argv[1] if sys.argv[3] != "insert": insert_data(conn) print("* benchmarking ...") if api == "executemany": with calc_time("executemany"): sql = sql_map[sys.argv[3]][1] cursor.executemany(sql, args) conn.commit() elif api == "execute_batch": with calc_time("execute_batch"): sql = sql_map[sys.argv[3]][1] execute_batch(cursor, sql, args) conn.commit() elif api == "prepare": with calc_time("execute_values"): cursor.execute(f"PREPARE test_stmt AS {sql_map[sys.argv[3]][2]}") if sys.argv[3] == "delete": execute_batch(cursor, "EXECUTE test_stmt (%s)", args) else: execute_batch(cursor, "EXECUTE test_stmt (%s, %s)", args) cursor.execute("DEALLOCATE test_stmt") conn.commit() elif api == "execute_values": with calc_time("execute_values"): sql = sql_map[sys.argv[3]][3] execute_values(cursor, sql, args) conn.commit() else: print(f"unknow api: {api}") if sys.argv[3] != "delete": cursor.execute("delete from t_psycopg2_benchmark") conn.commit() if __name__ == "__main__": main() ``` 本页内容