海南网站建设监理,wordpress阿里云配置文件,网站建设企业的市场分析,内贸在什么网站做这段代码使用了 Python 的 multiprocessing 模块来实现真正的并行处理#xff0c;绕过 Python 的全局解释器锁#xff08;GIL#xff09;限制#xff0c;从而在多核 CPU 上并发执行多个 SQL 语句。
from pyhive import hive
import multiprocessing# 建立连接
conn hive.… 这段代码使用了 Python 的 multiprocessing 模块来实现真正的并行处理绕过 Python 的全局解释器锁GIL限制从而在多核 CPU 上并发执行多个 SQL 语句。
from pyhive import hive
import multiprocessing# 建立连接
conn hive.Connection(hostlocalhost, port10000, usernameyour_username, passwordyour_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 multiprocessingpyhive: 这是用于连接和操作 Hive 数据库的 Python 库。hive.Connection 用于建立与 Hive 数据库的连接。multiprocessing: 这是 Python 的标准库用于创建和管理进程。通过 multiprocessing我们可以绕过 Python 的 GIL全局解释器锁限制实现真正的并行处理。 2. 建立数据库连接
conn hive.Connection(hostlocalhost, port10000, usernameyour_username, passwordyour_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 提供了一种方便的方式来并行化执行函数。其工作原理如下
创建进程池当你创建一个 Pool 对象时会启动多个工作进程数量可以指定或默认根据 CPU 核心数决定。任务分配当你调用 pool.map 时进程池会将任务在这里是 execute_sql 函数分配给空闲的工作进程。并行执行每个工作进程独立执行分配给它的任务互不干扰。结果收集pool.map 会收集所有工作进程的执行结果并按照原始任务列表的顺序返回结果。 为什么使用进程池而不是线程池
GIL 限制Python 的全局解释器锁GIL限制了多线程的并行执行能力尤其是在 CPU 密集型任务中多线程并不能充分利用多核 CPU。进程并行multiprocessing 模块通过创建多个进程来绕过 GIL 限制每个进程都有自己的 Python 解释器和内存空间因此可以实现真正的并行执行。适用场景 线程池适合 I/O 密集型任务例如等待数据库查询结果。进程池适合 CPU 密集型任务例如并行计算、数据处理等或者你需要绕过 GIL 限制时。 注意事项
数据库连接在多进程环境中每个进程都有自己的内存空间因此每个进程需要独立的数据库连接。在上述代码中每个进程都通过 conn.cursor() 创建了自己的游标。进程开销创建和销毁进程有一定的开销因此对于非常短小的任务进程池可能不会显著提高性能。在这种情况下可以考虑调整进程池的大小或使用其他优化手段。连接池如果你的程序需要频繁访问数据库可以考虑使用数据库连接池来复用数据库连接减少连接建立和关闭的开销。 总结
进程池通过 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 finset into {i}data.append(data_str)# 使用进程池并发执行with multiprocessing.Pool() as pool:pool.map(print_str, data)
