Python利用多进程将大量数据放入有限内存的教程

#import the necessary packages
import pandas as pd
import us
import numpy as np
from multiprocessing import Pool,cpu_count,Queue,Manager
 
# the data in one particular column was number in the form that horrible excel version
# of a number where '12000' is '12,000' with that beautiful useless comma in there.
# did I mention I excel bothers me"htmlcode">

#runs the function and puts the answers in the queue
def worker(row, q):
    ans = job_title_location_matcher(row[0],row[1])
    q.put(ans)
 
# this writes to the file while there are still things that could be in the queue
# this allows for multiple processes to write to the same file without blocking eachother
def listener(q):
  f = open(filename,'wb')
  while 1:
    m = q.get()
    if m =='kill':
        break
    f.write(','.join(m) + 'n')
    f.flush()
  f.close()
 
def main():
  #load all your data, then throw out all unnecessary tables/columns
  filename = 'skill_TEST_POOL.txt'
 
  #sets up the necessary multiprocessing tasks
  manager = Manager()
  q = manager.Queue()
  pool = Pool(cpu_count() + 2)
  watcher = pool.map_async(listener,(q,))
 
  jobs = []
  #titles_states is a dataframe of millions of job titles and states they were found in
  for i in titles_states.iloc:
    job = pool.map_async(worker, (i, q))
    jobs.append(job)
 
  for job in jobs:
    job.get()
  q.put('kill')
  pool.close()
  pool.join()
 
if __name__ == "__main__":
  main()




由于每个数据帧的大小都不同（总共约有100Gb），所以将所有数据都放入内存是不可能的。通过将最终的数据帧逐行写入内存，但从来不在内存中存储完整的数据帧。我们可以完成所有的计算和组合任务。这里的“标准方法”是，我们可以仅仅在“job_title_location_matcher”的末尾编写一个“write_line”方法，但这样每次只会处理一个实例。根据我们需要处理的职位/州的数量，这大概需要2天的时间。而通过multiprocessing，只需2个小时。
虽然读者可能接触不到本教程处理的任务环境，但通过multiprocessing，可以突破许多计算机硬件的限制。本例的工作环境是c3.8xl ubuntu ec2，硬件为32核60Gb内存（虽然这个内存很大，但还是无法一次性放入所有数据）。这里的关键之处是我们在60Gb的内存的机器上有效的处理了约100Gb的数据，同时速度提升了约25倍。通过multiprocessing在多核机器上自动处理大规模的进程，可以有效提高机器的利用率。也许有些读者已经知道了这个方法，但对于其他人，可以通过multiprocessing能带来非常大的收益。顺便说一句，这部分是skill assets in the job-market这篇博文的延续。