在一些文本的预处理和调研的时候，单进程太慢了，尤其是在几十万数据集的时候
解决如下：

基本框架

  1. import json
  2. from itertools import combinations
  3. import threading
  4. import time
  5. import datetime
  6. import multiprocessing as mp
  7. import copy
  8. import json
  9. import random
  10. random.seed(413)
  12. def Function(name,json_data):
  13.     result = {}
  14.     # Process Data
  15.     # Return reuslt in Dict format
  16.     return result
  18. def prepare_data():
  19.     input_data = []
  20.        # Prepare Data
  21.     # Single Sample is A Dict
  22.     return input_data
  24. def extract_result(results):
  25.     # Extract result and Display
  26.     # Result is a List According to Input result task division
  28. def multi_process_tag(target_data):
  29.     num_cores = int(mp.cpu_count())
  30.     print("本地计算机有: " + str(num_cores) + " 核心")
  31.     pool = mp.Pool(num_cores)
  33.     # Split the task based on CPU core in Computer
  34.     param_dict = {}
  35.     start = 0
  36.     end =  len(target_data)
  37.     step = int((end - start)/num_cores)
  38.     print("per Task Step: ",step)
  40.     # Construct Param Dict for multi-processing
  41.     for i in range(num_cores):
  42.         param_dict['task{}'.format(i)]= target_data[start:start+step]
  43.         start = start+step
  44.     param_dict['task{}'.format(num_cores)]= target_data[start:]
  45.     start_t = datetime.datetime.now()
  47.     # Run and get result
  48.     results = [pool.apply_async(Function, args=(name, param)) for name, param in param_dict.items()]
  49.     results = [p.get() for p in results]
  51.     end_t = datetime.datetime.now()
  52.     elapsed_sec = (end_t - start_t).total_seconds()
  53.     print("多进程计算 共消耗: " + "{:.2f}".format(elapsed_sec) + " 秒")
  54.     return results
  56. if __name__ ==  '__main__':
  57.     data = prepare_data()
  58.     results = multi_process_tag(data)
  59.     extract_result(results)

实例

2000条计算Oracle，原始大致10分钟左右，在8核I7上
用了多进程之后，一个核125条，不到3分钟解决

  1. import json
  2. from itertools import combinations
  3. from rouge import Rouge
  4. import threading
  5. import time
  6. import math
  7. import datetime
  8. import multiprocessing as mp
  9. from rouge import Rouge
  10. import copy
  11. import json
  12. import random
  13. random.seed(413)
  14. rouge = Rouge()
  16. def get_score(hyp,ref):
  17.     try:
  18.         temp_rouge = rouge.get_scores(hyp, ref)
  19.         cur_score = (temp_rouge[0]["rouge-1"]['f'] + temp_rouge[0]["rouge-2"]['f'] + temp_rouge[0]["rouge-l"]['f'])/3
  20.     except :
  21.         cur_score = 0
  22.     return cur_score
  24. def get_oracle(sent_list,summary):
  25.     Chosen_idx = []
  26.     best_score = 0
  27.     cal_count = 0 
  28.     while 1:
  29.         best_choice = -1
  30.         best_sub_score = 0
  31.         for i in range(len(sent_list)):
  32.             if i not in Chosen_idx and len(sent_list[i]) != 0 :
  33.                 cal_count += 1
  34.                 temp_chosen = copy.deepcopy(Chosen_idx)
  35.                 temp_chosen.append(i)
  36.                 temp_chosen_sents = [sent_list[i] for i in temp_chosen]
  37.                 #print(temp_chosen)
  38.                 #print(temp_chosen_sents)
  39.                 cur_score = get_score(" ".join(temp_chosen_sents),summary)
  40.                 cur_sub_score = cur_score - best_score
  41.                 if cur_sub_score > best_sub_score:
  42.                     best_sub_score = cur_sub_score
  43.                     best_choice = i
  44.         if best_choice == -1:
  45.             break
  46.         Chosen_idx.append(best_choice)
  47.         best_sents = [sent_list[i] for i in Chosen_idx]
  48.         best_score = get_score(" ".join(best_sents),summary)
  50.     best_sents = [sent_list[i] for i in Chosen_idx]
  51.     #print(len(sent_list))
  52.     #print(len(best_sents))
  53.     #print(cal_count)
  54.     try:
  55.         temp_rouge = rouge.get_scores(" ".join(best_sents), summary)
  56.     except :
  57.         return 0,0,0
  59.     return temp_rouge[0]["rouge-1"]['f'],temp_rouge[0]["rouge-2"]['f'],temp_rouge[0]["rouge-l"]['f']
  61. def Function(name,json_data):
  62.     result = {}
  63.     result['r1'] = []
  64.     result['r2'] = []
  65.     result['rl'] = []
  66.     for i in range(len(json_data)):
  67.         doc = json_data[i]['doc']
  68.         summary = json_data[i]['summary']
  69.         r1,r2,rl = get_oracle(doc,summary)
  70.         result['r1'].append(r1)
  71.         result['r2'].append(r2)
  72.         result['rl'].append(rl)
  75.     return result
  77. def prepare_data():
  78.     input_data = []
  79.     f = open('test.extract.source','r',encoding = 'utf-8')
  80.     f2 = open('test.target','r',encoding = 'utf-8')
  81.     f3 = open('QueryResult.txt','r',encoding = 'utf-8')
  82.     query = f3.readlines()
  83.     query = [[int (j) for j in i.strip().split()] for i in query]
  86.     summarys = f2.readlines()
  87.     summarys = [i.strip() for i in summarys]
  88.     import random
  89.     data_index = []
  90.     while len(data_index) < 2000:
  91.         random_index = random.randint(0,len(summarys)-1)
  92.         if random_index not in data_index:
  93.             data_index.append(random_index)
  95.     print(data_index[:10])
  96.     assert data_index[0] == 10455
  97.     lines = f.readlines()
  98.     ftrain = open('train.extract.source','r',encoding = 'utf-8')
  99.     assist_lines = ftrain.readlines()
  103.     for i in range(len(lines)):
  104.         data = lines[i].strip()
  105.         data_dict = json.loads(data)
  106.         doc = data_dict['text']
  108.         for j in query[i][:1]:
  109.             assist = assist_lines[j].strip()
  110.             assist_dict = json.loads(assist)
  111.             assist_doc = assist_dict['text']
  112.             doc = assist_doc + doc
  113.         temp_data = {}
  114.         temp_data['doc'] = doc
  115.         temp_data['summary'] = summarys[data_index[i]]
  117.         input_data.append(temp_data)
  118.     return input_data
  120. def extract_result(results):
  121.     total_samples = 0
  122.     Sum1 = 0
  123.     Sum2 = 0
  124.     SumL = 0
  125.     for i in results:
  126.         total_samples += len(i['r1'])
  127.         Sum1 += sum(i['r1'])
  128.         Sum2 += sum(i['r2'])
  129.         SumL += sum(i['rl'])
  131.     print(total_samples)
  132.     print(Sum1/total_samples)
  133.     print(Sum2/total_samples)
  134.     print(SumL/total_samples)
  136. def multi_process_tag(target_data):
  137.     num_cores = int(mp.cpu_count())
  138.     print("本地计算机有: " + str(num_cores) + " 核心")
  139.     pool = mp.Pool(num_cores)
  141.     param_dict = {}
  142.     start = 0
  143.     end =  len(target_data)
  144.     step = int((end - start)/num_cores)
  145.     print("per Task Step: ",step)
  146.     for i in range(num_cores):
  147.         param_dict['task{}'.format(i)]= target_data[start:start+step]
  148.         start = start+step
  149.     param_dict['task{}'.format(num_cores)]= target_data[start:]
  150.     start_t = datetime.datetime.now()
  151.     results = [pool.apply_async(Function, args=(name, param)) for name, param in param_dict.items()]
  152.     results = [p.get() for p in results]
  154.     end_t = datetime.datetime.now()
  155.     elapsed_sec = (end_t - start_t).total_seconds()
  156.     print("多进程计算 共消耗: " + "{:.2f}".format(elapsed_sec) + " 秒")
  157.     return results
  159. if __name__ ==  '__main__':
  160.     data = prepare_data()
  161.     results = multi_process_tag(data)
  162.     extract_result(results)