3

Помогите переделать функцию так, чтобы имплементировать многопоточную обработку данных и тем самым повысить скорость вычислений в N раз.

Я думаю что нужно менять ниже приведенный блок. Делить r_set_list на диапазоны и делать проверку на выполнение условия, и если хоть в одном диапазоне условие не выполняется, то прекращать проверку (убивая все остальные потоки). Как это красиво и кратко закодить пока не понимаю. И сложность сколько не в закодить и выбрать правильную стратегию распараллеливания. Затраты на организацию параллельной проверки внутри цикла, может потребовать больше ресурсов чем обработки в однопоточном режиме. Ведь код должен убивать все потоки, если хоть в одном из них выполнилось условие. А условие может выполниться на первом же элементе.

            f = True 
            for x in r_set_list:
                if (len(x & r_set)) > similar:
                    f = False
                    break

Исходный код

import pandas as pd 
import numpy as np 
import itertools as itr

df = pd.DataFrame(itr.combinations(range(1,26),6))
similar = 4
#print(df)

def FilterSimilarRecords(df, similar):
        r_set_list = list()
        bool_list = list()
        count = 0


        for r in df.values:
            count += 1
            if(count%100000 == 0):
                print("i:{}, r:{}".format(count, len(r_set_list)))

            r_set = set(r)

            f = True 
            for x in r_set_list:
                if (len(x & r_set)) > similar:
                    f = False
                    break

            if (f):
                bool_list.append(True)
                r_set_list.append(r_set)
            else:
                bool_list.append(False)

        return ~np.array(bool_list)

time_start = timer()
result = df[FilterSimilarRecords(df, similar)]
time_delta = timer() - time_start print('Execution time:', time_delta) print(result)

3 ответа 3

2

Я предположил следующие варианты решения:

  1. Оптимизировать однопоточный код.
  2. Разбить исполнение на потоки.
  3. Использовать JIT-компилятор.
  4. Использовать GPU.
  5. Использовать "более быстрые" языки программирования.

О том, что я не попробовал:

  1. Т.к. в конкретной задаче мы создает список уникальных значений и проходы по нему занимают основное время, то использование GPU может сильно ускорить задачу. Надеюсь, специалисты по Tensorflow, numba и пр. смогут написать пример быстрее, чем я разберусь. О проблематике подобных вычислений на GPU

  2. На C, Asm и пр. я не писал больше 10 лет и желания вспоминать - нет. ИМХО результат должен быть, как минимум, не сильно хуже варианта №3. Ну и т.к. речь в вопросе о Python, то этот вариант за рамками, я думаю.

Для тестирования я выделил из кода автора главное:

from timeit import default_timer as timer
import itertools as itr
import pandas as pd
import numpy as np

arr = list(itr.combinations(range(1,26),6))
similar = 4

def FilterSimilarRecords(df, similar):
        r_set_list = list()
        bool_list = list()

        for r in df.values:
            r_set = set(r)

            f = True 
            for x in r_set_list:
                if (len(x & r_set)) > similar:
                    f = False
                    break

            if (f):
                bool_list.append(True)
                r_set_list.append(r_set)
            else:
                bool_list.append(False)

        return ~np.array(bool_list)

time_start = timer()
df = pd.DataFrame(arr)
result = df[FilterSimilarRecords(df, similar)]
time_delta = timer() - time_start
print('Execution time:', time_delta)
print(result)

Результат: 72s [171468 rows x 6 columns]

1. Оптимизация однопоточного кода:

Исходя из тестов отдельных блоков задачи, наиболее быстрым однопоточным вариантом стал:

from timeit import default_timer as timer
import itertools as itr
import pandas as pd

arr = list(itr.combinations(range(1,26),6))
similar = 4

def FilterSimilarRecords(records, similar):
    unique = []
    nonunique = []
    # unique = deque()  # Аналогичный результат.
    # nonunique = deque()

    for r in records:
        r = set(r)
        for u in unique:
            if (len(u & r)) > similar:
                nonunique.append(r)
                break
        else:
            unique.append(r)
    return nonunique

time_start = timer()
filtered = pd.DataFrame(FilterSimilarRecords(arr, similar))
time_delta = timer() - time_start
print('Execution time:', time_delta)
print(filtered)

Результат: 65s [171468 rows x 6 columns]

2. Разбить исполнение на потоки:

Многопоточность - не помогает. Специфика задачи такова, что процессор остается не нагружен, а узким местом является доступ к памяти. Соответственно, мы наблюдаем снижение производительности из-за накладных расходов на организацию многопоточности.

Исходя из тестов отдельных блоков задачи, наиболее быстрым многопоточным вариантом стал:

from multiprocessing.dummy import Pool
from timeit import default_timer as timer
import itertools as itr
import pandas as pd

arr = list(itr.combinations(range(1,26),6))

class FilterSimilarRecords:
    pool = Pool(16)
    unique = []
    nonunique = []
    similar = 4
    is_unique = True

    def run(self, arr):
        for a in arr:
            self.is_unique = True
            self.record = set(a)
            self.pool.map(self.apply, self.unique)
            if self.is_unique:
                self.unique.append(self.record)
            else:
                self.nonunique.append(self.record)

        return self.nonunique

    def apply(self, u):
        if (len(u & self.record)) > self.similar:
            self.is_unique = False
            raise StopIteration


time_start = timer()
result = FilterSimilarRecords().run(arr)
filtered = pd.DataFrame(result)
time_delta = timer() - time_start
print('Execution time:', time_delta)
print(filtered)

Результат: 392s [171468 rows x 6 columns] C try, except - результат аналогичен.

3. Использовать JIT-компилятор:

Наиболее быстрый из протестированних на CPU. np.array - медленнее. Буду рад, если кто-нибудь реализует на GPU. У меня "с наскока" ничего хорошего не вышло.

from timeit import default_timer as timer
import itertools as itr
import numpy as np
import pandas as pd
import numba

arr = np.array(list(itr.combinations(range(1,26),6)))
similar = 4

@numba.njit(target='cpu')
def FilterSimilarRecords(arr, similar):
    unique = list()
    nonunique = list()
    for a in range(arr.shape[0]):
        r = set(arr[a])
        for u in range(len(unique)):
            if (len(unique[u] & r)) > similar:
                nonunique.append(arr[a])
                break
        else:
            unique.append(r)
    return nonunique

time_start = timer()
filtered = pd.DataFrame(FilterSimilarRecords(arr, similar))
time_delta = timer() - time_start
print('Execution time:', time_delta)
print(filtered)

Результат: 50s [171468 rows x 6 columns]

10
  • Интересно решение на GPU. Есть идея оптимизировать алгоритм для 1) и 3) случаев. Суммировать элементы строки и делать поиск по диапазону сумм. На днях выложу код. 5 авг 2019 в 5:35
  • Меня интересовала оптимизация для "общего случая". Т.е. с возможностью использования в иных расчетах. 5 авг 2019 в 5:42
  • Для "общего случая" сложновато будет. В вашем случае в строке могут быть не уникальные значения? В моем только уникальные. В задаче самая тяжелая часть выпадает на (len(u & r)) > similar . Я думал переводить u и r в int128 (в моей задаче макс знач. элемента = 80). В теории побитовая операция & быстрая, но вот подсчет кол-во бит будет медленным. Поэтому оставил эту затею. 5 авг 2019 в 17:30
  • Под "общим случаем", я понимал применение быстрой функции к длинному списку. 5 авг 2019 в 17:35
  • Кстати for a in range(arr.shape[0]) нельзя параллелить, поскольку тогда может нарушиться логика выборки. Данные должны сканироваться и обрабатываться последовательно. А вы можете показать GPU версию пусть даже медленную? 5 авг 2019 в 17:37
0

Улучшенная многопоточная версию - 2. Разбить исполнение на потоки

Тесты делались на ноутбуке. CPU i5-6200 2.30GHZ

Исходная версия: Execution time: 559.0649321000001

Улучшенная версия: Execution time: 268.8355488999996

from multiprocessing.dummy import Pool
from timeit import default_timer as timer
import itertools as itr
import pandas as pd
import gmpy2

df = pd.DataFrame(list(itr.combinations(range(1,26),6)))


class FilterSimilarRecords:
    pool = Pool(16)
    unique = []
    unique_key = []    
    is_unique = True

    def run(self, df, similar):
        dct = df.T.to_dict('list')
        dct_bin = dict(map(lambda item: (item[0], self.ListIntToBigInt(item[1])), dct.items()))

        lst_x = 0
        lst_y = 0
        lst = []
        lst_size = 10000
        len_dct = len(dct) 
        unique = []
        unique_key = []        
        key = 0
        self.similar = similar

        while key < len_dct:
            self.record = dct_bin[key]
            self.is_unique = True
            self.pool.map(self.apply_lst, lst)
            if self.is_unique:
                self.unique.append(self.record)
                self.unique_key.append(key)

                if(lst_x > lst_size):
                    lst_x = 0
                    lst_y += 1
                    lst.append([])
                elif (lst_x == 0):
                    lst.append([])

                l = lst[lst_y]
                l.append(self.record)
                lst_x += 1                

            key += 1

        return df[~df.index.isin(self.unique_key)]

    def apply_lst(self, u):
        if(next(filter(lambda x: (gmpy2.popcount(self.record & x)) > self.similar, u)) != None):
            self.is_unique = False
            raise StopIteration            


    def ListIntToBigInt(self, lst):
        return sum(2**i for i in lst)


time_start = timer()
result = FilterSimilarRecords().run(df, 4)
time_delta = timer() - time_start
print('Execution time:', time_delta)
print(result)
0

Улучшенная однопоточная версию - 1. Оптимизация однопоточного кода

Тесты делались на ноутбуке. CPU i5-6200 2.30GHZ

Исходная версия: Execution time: 184.5349679

Улучшенная версия: Execution time: 107.32932070000004

from timeit import default_timer as timer
import itertools as itr
import pandas as pd
import gmpy2

def ListIntToBigInt(lst):
    return sum(2**i for i in lst)

df = pd.DataFrame(list(itr.combinations(range(1,26),6)))

similar = 4

def FilterSimilarRecords(df, similar):
    dct = df.T.to_dict('list')
    dct_bin = dict(map(lambda item: (item[0], ListIntToBigInt(item[1])), dct.items()))

    unique = []
    unique_key = []
    key = 0

    len_dct = len(dct)

    while key < len_dct:
        r = dct_bin[key]
        for u in unique:
            if (gmpy2.popcount(u & r)) > similar:
                break
        else:
            unique.append(r)
            unique_key.append(key)

        key += 1

    return df[~df.index.isin(unique_key)]

time_start = timer()
filtered = FilterSimilarRecords(df, similar)
time_delta = timer() - time_start
print('Execution time:', time_delta)
print(filtered)

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.