Python并行处理技巧:利用多线程加速For循环
本文探讨如何在Python中通过多线程技术加速耗时的For循环操作。针对大规模数据(如千万级记录)处理场景,通过示例展示如何将串行处理改为并行处理,利用multiprocessing.dummy模块中的ThreadPool实现并发执行,显著减少运行时间,并提供具体代码实现与性能对比结果。
大规模数据处理的串行瓶颈
以某城市出租车数据为例,单日记录量高达33210000条。若需将每辆车的数据单独提取并保存至独立文件,该如何高效实现?
基本思路如下:
遍历全部33210000条记录,将每条数据写入对应的专属文件中。
然而,面对3000多万条数据,串行逐个循环处理耗时极长。实测中,处理60万数据耗时2小时,推算处理全部3000万数据需约100小时(即4-5天)。此外,还需确保设备连续运行五天且无崩溃风险。
并行For循环的设计思路
因此,引入并行处理For循环的技巧成为必要方案:
由于3000万数据直接存入单个CSV文件会导致文件过大无法打开,故使用split工具将原CSV分割为53个文件,每个文件约60万条记录。
原方案逻辑为:读取文件夹,获取所有60万条记录的CSV文件列表,再逐一处理。这本质上仍是33210000次循环。并行For循环的核心在于同时处理多个60万条记录的CSV文件,从而成倍降低总耗时。
并行For循环的设计灵感源自以下方法:
传统串行处理方式类似如下代码结构:
words = ['apple', 'bananan', 'cake', 'dumpling']
for word in words:
print word
并行For循环的实现方式如下:
from multiprocessing.dummy import Pool as ThreadPool
items = list()
pool = ThreadPool()
pool.map(process, items)
pool.close()
pool.join()
其中,process为实际执行处理的函数。
Python多线程并行处理实现
具体实例代码如下:
# -*- coding: utf-8 -*-
import time
from multiprocessing.dummy import Pool as ThreadPool
def process(item):
print('正在并行for循环')
print(item)
time.sleep(5)
items = ['apple', 'bananan', 'cake', 'dumpling']
pool = ThreadPool()
pool.map(process, items)
pool.close()
pool.join()
补充知识:在Python3中,可使用多线程替代传统For循环以提升程序运行速度。
性能优化对比与结果分析
优化前后的代码对比如下:
from git_tools.git_tool import get_collect_projects, QQNews_Git
from threading import Thread, Lock
import datetime
base_url = "http://git.xx.com"
project_members_commits_lang_info = {}
lock = Lock()
threads = []
'''
Author:zenkilan
'''
def count_time(func):
def took_up_time(*args, **kwargs):
start_time = datetime.datetime.now()
ret = func(*args, **kwargs)
end_time = datetime.datetime.now()
took_up_time = (end_time - start_time).total_seconds()
print(f"{func.__name__} execution took up time:{took_up_time}")
return ret
return took_up_time
def get_project_member_lang_code_lines(git, member, begin_date, end_date):
global project_members_commits_lang_info
global lock
member_name = member["username"]
r = git.get_user_info(member_name)
if not r["id"]:
return
user_commits_lang_info = git.get_commits_user_lang_diff_between(r["id"], begin_date, end_date)
if len(user_commits_lang_info) == 0:
return
lock.acquire()
project_members_commits_lang_info.setdefault(git.project, dict())
project_members_commits_lang_info[git.project][member_name] = user_commits_lang_info
lock.release()
def get_project_lang_code_lines(project, begin_date, end_date):
global threads
git = QQNews_Git(project[1], base_url, project[0])
project_members = git.get_project_members()
if len(project_members) == 0:
return
for member in project_members:
thread = Thread(target=get_project_member_lang_code_lines, args=(git, member, begin_date, end_date))
threads.append(thread)
thread.start()
@count_time
def get_projects_lang_code_lines(begin_date, end_date):
"""
获取项目代码行语言相关统计——新方法(提升效率)
应用多线程替代for循环
并发访问共享外部资源
:return:
"""
global project_members_commits_lang_info
global threads
for project in get_collect_projects():
thread = Thread(target=get_project_lang_code_lines, args=(project, begin_date, end_date))
threads.append(thread)
thread.start()
@count_time
def get_projects_lang_code_lines_old(begin_date, end_date):
"""
获取项目代码行语言相关统计——老方法(耗时严重)
使用最基本的思路进行编程
双层for循环嵌套并且每层都包含耗时操作
:return:
"""
project_members_commits_lang_info = {}
for project in get_collect_projects():
git = QQNews_Git(project[1], base_url, project[0])
project_members = git.get_project_members()
user_commits_lang_info_dict = {}
if len(project_members) == 0:
continue
for member in project_members:
member_name = member["username"]
r = git.get_user_info(member_name, debug=False)
if not r["id"]:
continue
try:
user_commits_lang_info = git.get_commits_user_lang_diff_between(r["id"], begin_date, end_date)
if len(user_commits_lang_info) == 0:
continue
user_commits_lang_info_dict[member_name] = user_commits_lang_info
project_members_commits_lang_info[git.project] = user_commits_lang_info_dict
except:
pass
return project_members_commits_lang_info
def test_results_equal(resultA, resultB):
"""
测试方法
:param resultA:
:param resultB:
:return:
"""
print(resultA)
print(resultB)
assert len(str(resultA)) == len(str(resultB))
if __name__ == '__main__':
from git_tools.config import begin_date, end_date
get_projects_lang_code_lines(begin_date, end_date)
for t in threads:
t.join()
old_result = get_projects_lang_code_lines_old(begin_date, end_date)
test_results_equal(old_result, project_members_commits_lang_info)
老方法中,外层For循环和内层For循环均包含耗时操作:
1)git.get_project_members()
2)git.get_user_info(member_name, debug=False)
优化分两步进行,先内层后外层或先外层后内层均可。采用多线程替换For循环,并发共享外部资源,并通过加锁机制避免写入冲突。
测试结果验证通过,函数运行时间装饰器显示结果如下(单位:秒):
get_projects_lang_code_lines execution took up time:1.85294
get_projects_lang_code_lines_old execution took up time:108.604177
速度提升约58倍。
以上即为提高Python中For循环效率的完整分享,供读者参考,欢迎支持本站。
免责声明:本文内容仅供参考,不代表本站观点。转载或引用请注明出处。









