快捷導(dǎo)航

如何提高python 中for循環(huán)的效率

更新時間：2020年04月15日 09:11:36 作者：winyn_

這篇文章主要介紹了如何提高python 中for循環(huán)的效率，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧

對于某個城市的出租車數(shù)據(jù)，一天就有33210000條記錄，如何將每輛車的數(shù)據(jù)單獨拎出來放到一個專屬的文件中呢？

思路很簡單：

就是循環(huán)33210000條記錄，將每輛車的數(shù)據(jù)搬運到它該去的文件中。

但是對于3000多萬條數(shù)據(jù)，一個一個循環(huán)太消耗時間，我花了2個小時才搬運了60萬數(shù)據(jù)，算算3000萬我需要花費100個小時，也就需要4-5天。并且還需要保證這五天全天開機(jī)，不能出現(xiàn)卡機(jī)的事故。

因此，需要使用并行進(jìn)行for循環(huán)的技巧：

由于3000萬數(shù)據(jù)放到csv中導(dǎo)致csv打不開，因此我就把一個csv通過split軟件將其切分成每份60萬，共53個csv。

我原來的思路是讀取文件夾，獲取由每一個60萬的csv文件組成的列表，再分別對每一個60萬的csv進(jìn)行處理。實質(zhì)上還是循環(huán)33210000次，并行for循環(huán)就是同時處理幾個60萬的csv文件，就能成倍的減少時間消耗。

并行進(jìn)行for循環(huán)是受下面的方法啟發(fā)：

我之前的做法類似這樣：

  words = ['apple', 'bananan', 'cake', 'dumpling']
  for word in words:
    print word

并行for循環(huán)類似這樣：

from multiprocessing.dummy import Pool as ThreadPool
items = list()
pool = ThreadPool()
pool.map(process, items)
pool.close()
pool.join()

其中,process是進(jìn)行處理的函數(shù)

實例代碼如下：

# -*- coding: utf-8 -*-
import time
from multiprocessing.dummy import Pool as ThreadPool
def process(item):
  print('正在并行for循環(huán)')
  print(item)
  time.sleep(5)
items = ['apple', 'bananan', 'cake', 'dumpling']
pool = ThreadPool()
pool.map(process, items)
pool.close()
pool.join()

補(bǔ)充知識：Python3用多線程替代for循環(huán)提升程序運行速度

優(yōu)化前后新老代碼如下：

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):
  """
  獲取項目代碼行語言相關(guān)統(tǒng)計——新方法（提升效率）
  應(yīng)用多線程替代for循環(huán)
  并發(fā)訪問共享外部資源
  :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):
  """
  獲取項目代碼行語言相關(guān)統(tǒng)計——老方法（耗時嚴(yán)重）
  使用最基本的思路進(jìn)行編程
  雙層for循環(huán)嵌套并且每層都包含耗時操作
  :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循環(huán)和內(nèi)層for循環(huán)里均存在耗時操作：

1）git.get_project_members()

2）git.get_user_info(member_name, debug=False)

分兩步來優(yōu)化，先里后外或先外后里都行。用多線程替換for循環(huán)，并發(fā)共享外部資源，加鎖避免寫沖突。

測試結(jié)果通過，函數(shù)運行時間裝飾器顯示（單位秒）：