当前位置： 首页 > news >正文

python 使用进程池并发执行 SQL 语句

news 2025/6/19 10:46:52

        这段代码使用了 Python 的 multiprocessing 模块来实现真正的并行处理，绕过 Python 的全局解释器锁（GIL）限制，从而在多核 CPU 上并发执行多个 SQL 语句。

from pyhive import hive
import multiprocessing# 建立连接
conn = hive.Connection(host="localhost", port=10000, username="your_username", password="your_password")# SQL 语句列表
sql_statements = ["INSERT INTO table1 VALUES (1, 'value1')","INSERT INTO table1 VALUES (2, 'value2')","INSERT INTO table1 VALUES (3, 'value3')"
]# 定义执行函数
def execute_sql(sql):with conn.cursor() as cursor:cursor.execute(sql)# 确保多进程代码只在主进程中执行
if __name__ == '__main__':# 使用进程池并发执行with multiprocessing.Pool() as pool:pool.map(execute_sql, sql_statements)# 关闭连接conn.close()

1. 导入模块

from pyhive import hive
import multiprocessing
  • pyhive: 这是用于连接和操作 Hive 数据库的 Python 库。hive.Connection 用于建立与 Hive 数据库的连接。
  • multiprocessing: 这是 Python 的标准库，用于创建和管理进程。通过 multiprocessing，我们可以绕过 Python 的 GIL（全局解释器锁）限制，实现真正的并行处理。

2. 建立数据库连接

conn = hive.Connection(host="localhost", port=10000, username="your_username", password="your_password")
  • 这里我们使用 hive.Connection 建立一个到 Hive 数据库的连接。
  • 参数：
    • host: HiveServer2 的主机地址，通常是 localhost 或 HiveServer2 运行的服务器 IP。
    • port: HiveServer2 的端口号，默认是 10000。
    • username: 连接 Hive 使用的用户名。
    • password: 连接 Hive 使用的密码。

这个连接对象 conn 将在后续的代码中用于创建游标（cursor），并通过游标执行 SQL 语句。

3. 定义 SQL 语句列表

sql_statements = ["INSERT INTO table1 VALUES (1, 'value1')","INSERT INTO table1 VALUES (2, 'value2')","INSERT INTO table1 VALUES (3, 'value3')"
]
  • 这里定义了一个包含多个 SQL 语句的列表 sql_statements。每个语句都是一个插入操作，将数据插入到 Hive 表 table1 中。
  • 你可以根据实际需求修改这些 SQL 语句。

4. 定义执行函数

def execute_sql(sql):with conn.cursor() as cursor:cursor.execute(sql)
  • execute_sql 函数是用于执行单个 SQL 语句的函数。
  • with conn.cursor() as cursor：为当前数据库连接创建一个游标对象 cursor，这个游标用于执行 SQL 语句。
    • cursor.execute(sql)：执行传入的 SQL 语句。
  • 这个函数会被进程池中的每个进程调用，每个进程都会独立执行一个 SQL 语句。

5. 使用进程池并发执行

with multiprocessing.Pool() as pool:pool.map(execute_sql, sql_statements)
  • multiprocessing.Pool()：创建一个进程池。进程池可以管理一组工作进程，并将任务分配给这些进程。
    • 默认情况下，Pool() 会根据系统的 CPU 核心数创建相应数量的工作进程。
    • 你可以通过参数指定池中的进程数量，例如 Pool(4) 表示创建 4 个工作进程。
  • pool.map(execute_sql, sql_statements)：
    • pool.map 方法会将 execute_sql 函数应用到 sql_statements 列表中的每个元素上。
    • pool.map 方法会自动将 SQL 语句列表分配给进程池中的工作进程，每个进程独立执行一个 SQL 语句。
    • 这个过程是并行的，多个进程可以同时执行不同的 SQL 语句，从而提高执行效率。

6. 关闭数据库连接

conn.close()
  • 在所有 SQL 语句执行完毕后，我们关闭数据库连接，释放资源。

进程池的工作原理

multiprocessing.Pool 提供了一种方便的方式来并行化执行函数。其工作原理如下：

  1. 创建进程池：当你创建一个 Pool 对象时，会启动多个工作进程（数量可以指定，或默认根据 CPU 核心数决定）。
  2. 任务分配：当你调用 pool.map 时，进程池会将任务（在这里是 execute_sql 函数）分配给空闲的工作进程。
  3. 并行执行：每个工作进程独立执行分配给它的任务，互不干扰。
  4. 结果收集：pool.map 会收集所有工作进程的执行结果，并按照原始任务列表的顺序返回结果。

为什么使用进程池而不是线程池？

  1. GIL 限制：Python 的全局解释器锁（GIL）限制了多线程的并行执行能力，尤其是在 CPU 密集型任务中，多线程并不能充分利用多核 CPU。
  2. 进程并行：multiprocessing 模块通过创建多个进程来绕过 GIL 限制，每个进程都有自己的 Python 解释器和内存空间，因此可以实现真正的并行执行。
  3. 适用场景：
    • 线程池：适合 I/O 密集型任务（例如，等待数据库查询结果）。
    • 进程池：适合 CPU 密集型任务（例如，并行计算、数据处理等），或者你需要绕过 GIL 限制时。

注意事项

  1. 数据库连接：在多进程环境中，每个进程都有自己的内存空间，因此每个进程需要独立的数据库连接。在上述代码中，每个进程都通过 conn.cursor() 创建了自己的游标。
  2. 进程开销：创建和销毁进程有一定的开销，因此对于非常短小的任务，进程池可能不会显著提高性能。在这种情况下，可以考虑调整进程池的大小或使用其他优化手段。
  3. 连接池：如果你的程序需要频繁访问数据库，可以考虑使用数据库连接池来复用数据库连接，减少连接建立和关闭的开销。

总结

  • 进程池：通过 multiprocessing.Pool 实现，可以绕过 Python 的 GIL 限制，实现真正的并行处理。
  • 适用场景：适合 CPU 密集型任务或需要并行执行多个独立任务的场景。
  • 代码结构：
    • 建立数据库连接。
    • 定义 SQL 语句列表。
    • 定义执行函数 execute_sql。
    • 使用进程池并发执行 SQL 语句。
    • 关闭数据库连接。

通过这种方式，你可以充分利用多核 CPU 的优势，并发执行多个 SQL 语句，从而提高程序的执行效率。

解决多进程报错

你遇到的错误是 RuntimeError，这是因为你在使用 multiprocessing 时没有正确地保护代码的入口点。具体来说，在 Windows 系统上（以及其他非 fork 的启动方式），你必须将多进程相关的代码放在 if __name__ == '__main__': 语句块中，以避免子进程在启动时重新导入主模块并执行不必要的代码。

错误原因：

在 Windows 系统中，Python 的 multiprocessing 模块使用 spawn 启动子进程，这意味着子进程会重新导入当前脚本。如果不加以保护，子进程会再次执行主模块中的代码，导致递归创建进程并抛出错误。

解决方案：

你需要将多进程相关的代码放在 if __name__ == '__main__': 语句块中，确保只有主进程会执行这些代码，而子进程不会。

修改后的代码：

 

python
 

 

import multiprocessingdata = ["1","2","3"
]# 定义执行函数
def print_str(data):print(data)# 确保多进程代码只在主进程中执行
if __name__ == '__main__':# 使用进程池并发执行with multiprocessing.Pool() as pool:pool.map(print_str, data)
 

解释：
 

  • if __name__ == '__main__': 确保了只有在直接运行当前脚本时，才会执行其中的多进程代码。子进程不会执行这个代码块，从而避免了递归创建进程的问题。
  • 在 Windows 系统上，这是使用 multiprocessing 时必须遵循的惯用写法。
 

其他注意事项：
 

  • 如果你打算将脚本打包成可执行文件（例如使用 pyinstaller），你还需要调用 multiprocessing.freeze_support()，不过在大多数脚本运行的情况下，这个调用不是必须的。
 

例如：
 

 

python
 

 

if __name__ == '__main__':multiprocessing.freeze_support()  # 如果需要打包成可执行文件，可以加上这行with multiprocessing.Pool() as pool:pool.map(print_str, data)
 

参考文档：
 

你可以参考 Python 官方文档中关于 multiprocessing 的部分，了解更多关于安全导入主模块的信息：
 

  • multiprocessing — Process-based parallelism — Python 3.13.0 documentation
 

 

执行sql 简单示例
 

import multiprocessingdata = [  ]# 定义执行函数
def print_str(data):print(data)# 确保多进程代码只在主进程中执行
if __name__ == '__main__':data2 = ["1","2","3"]for i in data2:data_str = f"""inset into {i}"""data.append(data_str)# 使用进程池并发执行with multiprocessing.Pool() as pool:pool.map(print_str, data)

 






http://www.lryc.cn/news/476549.html






相关文章：


  • 
我也谈AI

    • 

  • 
算法妙妙屋-------1.递归的深邃回响：二叉树的奇妙剪枝

    • 

  • 
编写第一个 Appium 测试脚本：从安装到运行！

    • 

  • 
mysql查表相关练习

    • 

  • 
airtest+poco多脚本、多设备批处理运行测试用例自动生成测试报告

    • 

  • 
Prometheus套装部署到K8S+Dashboard部署详解

    • 

  • 
python使用pymysql

    • 

  • 
Vue3 + TypeScript 组件和文件命名规范及 setup 导入顺序规范

    • 

  • 
netty之处理连接源码分析

    • 

  • 
Dockerfile文件编写

    • 

  • 
Oracle SQL 使用 ROWNUM 分页查询速度太慢的问题及解决方案！

    • 

  • 
Django3 + Vue.js 前后端分离书籍添加项目Web开发实战

    • 

  • 
楼梯区域分割系统：Web效果惊艳

    • 

  • 
Day10加一

    • 

  • 
UTF-8简介

    • 

  • 
基于Openwrt系统架构，实现应用与驱动的实例。

    • 

  • 
SQL进阶技巧：如何利用三次指数平滑模型预测商品零售额？

    • 

  • 
HTB：Cicada[WriteUP]

    • 

  • 
小张求职记四

    • 

  • 
适用于 c++ 的 wxWidgets框架源码编译SDK-windows篇

    • 

  • 
flink 内存配置（二）：设置TaskManager内存

    • 

  • 
【C++ 算法进阶】算法提升八

    • 

  • 
阿里云实时数据仓库HologresFlink

    • 

  • 
生成式语言模型的文本生成评价指标(从传统的基于统计到现在的基于语义)

    • 

  • 
【网安案例学习】暴力破解攻击（Brute Force Attack）

    • 

  • 
时间序列预测（十八）——实现配置管理和扩展命令行参数解析器

    • 

  • 
Vue问题汇总解决

    • 

  • 
Spark学习

    • 

  • 
一些小细节代码笔记汇总

    • 

  • 
L4.【LeetCode笔记】链表题的VS平台调试代码

    •