Блог

У меня есть функция, которая возвращает словарь. Функция работает путем вычисления значений на основе массива в фрейме данных.

Фрейм данных содержит около 1000 000 строк и выглядит следующим образом:

                   col1                  
row1         [2, 3, 44, 89.6,...]           
row2         [10, 4, 33.3, 1.11,...]
row3         [3, 4, 3, 2.6, 5.9, 8, 10,...]  
  

Моя функция принимает каждый массив в каждой строке, выполняет некоторые вычисления и возвращает словарь на основе этих вычислений. Однако это происходит очень медленно. Я ценю необходимость просеивания большого количества данных, но есть ли способ, которым я могу улучшить скорость?

Проблемы с фреймом данных большой длины. Каждый массив может содержать более 100 значений. Колеблется примерно в пределах 10-80.

Мой код выглядит следующим образом:

 list1 = []

for i in df.itertuples():
    list1.append(list(function(i.data).values()))
  

Идея здесь в том, что я перебираю каждую строку в ‘df’, применяю свою функцию к столбцу ‘data’ и добавляю результаты в список ‘list1’.

Объяснена функция

Моя функция вычисляет некоторые довольно простые вещи. Он принимает массив в качестве параметра и вычисляет данные на основе этого массива, например, его длину, среднее значение в массиве, минимальное и максимальное значение массива. Я вычисляю 8 значений и сохраняю их в словаре. Последнее, что делает моя функция, это просматривает эти вычисленные значения и добавляет окончательный ключ в словарь в виде логического значения.

Как я уже говорил в комментариях, если ваша функция дорогостоящая (сокращение каждой строки — это трудоемкая часть вашего кода), то первым шагом является использование multiprocessing , потому что ее легко протестировать.

Вот что вы могли бы попробовать:

 import time
from multiprocessing import Pool

def f(x):
  time.sleep(10*10**-6) # Faking complex computation
  return x

def seq_test(input_array):
  return list(map(f, input_array))

def par_test(input_array):
  pool = Pool(8)  #  "nproc --all" or "sysctl -n hw.ncpu" on osx
  return pool.map(f, input_array)

def run_test(test_function):
  test_size = 10*10**4
  test_input = [i for i in range(test_size)]

  t0 = time.time()
  result = test_function(test_input)
  t1 = time.time()

  print(f"{test_function.__name__}: {t1-t0:.3f}s")

run_test(seq_test)
run_test(par_test)
  

На моей машине параллельная версия работает примерно в 7 раз быстрее (довольно близко к коэффициенту 8, на который мы могли надеяться):

 seq_test: 2.131s
par_test: 0.300s
  

Если этого недостаточно, следующим шагом будет написание функции f на другом языке, еще раз, что кажется более простым здесь, так это использовать Cython. Но для обсуждения этого нам нужно посмотреть, что находится внутри вашей функции.

Я предлагаю изменить формат ваших данных следующим образом:

 print (df)
                            col1
row1            [2, 3, 44, 89.6]
row2         [10, 4, 33.3, 1.11]
row3  [3, 4, 3, 2.6, 5.9, 8, 10]

from itertools import chain

df = pd.DataFrame({
    'idx' : df.index.repeat(df['col1'].str.len()),
    'col1' : list(chain.from_iterable(df['col1'].tolist()))
})
print (df)
     idx   col1
0   row1   2.00
1   row1   3.00
2   row1  44.00
3   row1  89.60
4   row2  10.00
5   row2   4.00
6   row2  33.30
7   row2   1.11
8   row3   3.00
9   row3   4.00
10  row3   3.00
11  row3   2.60
12  row3   5.90
13  row3   8.00
14  row3  10.00
  

А затем агрегируйте ваши данные:

 df1 = df.groupby('idx')['col1'].agg(['sum','mean','max','min'])
print (df1)
         sum       mean   max   min
idx                                
row1  138.60  34.650000  89.6  2.00
row2   48.41  12.102500  33.3  1.11
row3   36.50   5.214286  10.0  2.60