Операции параллелизации в python

Question

Я заполняю матрицу, используя условный поиск из файла. Файл чрезвычайно велик (25 000 000 записей) и сохраняется как файл данных («файл»). Каждая операция строки матрицы (поиск) не зависит от другой. Есть ли способ распараллеливать этот процесс?

Я работаю в пандах и питоне. Мой нынешний подход голый наивный.

for r in row: 
    for c in column: 
     num=file[(file['Unique_Inventor_Number']==r) & file['AppYearStr']==c)]['Citation'].tolist() 
     num = len(list(set(num))) 
     d.set_value(r, c, num) 

Answer 1

За 2,5 миллиона записей вы должны быть в состоянии сделать

res = file.groupby(['Unique_Inventor_Number', 'AppYearStr']).Citation.nunique() 

Матрица должна быть доступна в

res.unstack(level=1).fillna(0).values 

Я не уверен, если это самый быстрый, но должна быть значительно быстрее, чем реализации

Answer 2

[EDIT] Как Roland упоминалось в комментариях, в стандартной реализации в Python, этот пост не предлагает никакого решения для повышения производительности процессоров.

В стандартной реализации Python потоки на самом деле не улучшают производительность задач, связанных с CPU. Существует «Global Interpreter Lock», который предусматривает, что только один поток за раз может выполнять байт-код Python. Это было сделано, чтобы снизить сложность управления памятью.

Вы пытались использовать разные темы для различных функций?

Предположим, вы отделили свой файл данных от столбцов и создали несколько потоков. Затем вы назначаете каждый поток для применения функции к столбцу. Если у вас есть достаточно вычислительной мощности, вы можете быть в состоянии получить много времени:

from threading import Thread 
import pandas as pd 
import numpy as np 
from queue import Queue 
from time import time 

# Those will be used afterwards 
N_THREAD = 8 
q = Queue() 
df2 = pd.DataFrame() # The output of the script 

# You create the job that each thread will do 
def apply(series, func): 
    df2[series.name] = series.map(func) 


# You define the context of the jobs 
def threader(): 
    while True: 
     worker = q.get() 
     apply(*worker) 
     q.task_done() 

def main(): 

    # You import your data to a pandas dataframe 
    df = pd.DataFrame(np.random.randn(100000,4), columns=['A', 'B', 'C', 'D']) 

    # You create the functions you will apply to your columns 
    func1 = lambda x: x<10 
    func2 = lambda x: x==0 
    func3 = lambda x: x>=0 
    func4 = lambda x: x<0 
    func_rep = [func1, func2, func3, func4] 

    for x in range(N_THREAD): # You create your threads  
     t = Thread(target=threader) 
     t.start() 

    # Now is the tricky part: You enclose the arguments that 
    # will be passed to the function into a tuple which you 
    # put into a queue. Then you start the job by "joining" 
    # the queue 
    for i, func in enumerate(func_rep): 
     worker = tuple([df.iloc[:,i], func]) 
     q.put(worker) 

    t0 = time() 
    q.join() 
    print("Entire job took: {:.3} s.".format(time() - t0)) 

if __name__ == '__main__': 
    main()