Применение функции Python к сгруппированной в DataFrame группе Pandas - какой наиболее эффективный подход для ускорения вычислений?

9

Я имею дело с довольно большим Pandas DataFrame - мой набор данных похож на следующую dfнастройку:

import pandas as pd
import numpy  as np

#--------------------------------------------- SIZING PARAMETERS :
R1 =                    20        # .repeat( repeats = R1 )
R2 =                    10        # .repeat( repeats = R2 )
R3 =                541680        # .repeat( repeats = [ R3, R4 ] )
R4 =                576720        # .repeat( repeats = [ R3, R4 ] )
T  =                 55920        # .tile( , T)
A1 = np.arange( 0, 2708400, 100 ) # ~ 20x re-used
A2 = np.arange( 0, 2883600, 100 ) # ~ 20x re-used

#--------------------------------------------- DataFrame GENERATION :
df = pd.DataFrame.from_dict(
         { 'measurement_id':        np.repeat( [0, 1], repeats = [ R3, R4 ] ), 
           'time':np.concatenate( [ np.repeat( A1,     repeats = R1 ),
                                    np.repeat( A2,     repeats = R1 ) ] ), 
           'group':        np.tile( np.repeat( [0, 1], repeats = R2 ), T ),
           'object':       np.tile( np.arange( 0, R1 ),                T )
           }
        )

#--------------------------------------------- DataFrame RE-PROCESSING :
df = pd.concat( [ df,
                  df                                                  \
                    .groupby( ['measurement_id', 'time', 'group'] )    \
                    .apply( lambda x: np.random.uniform( 0, 100, 10 ) ) \
                    .explode()                                           \
                    .astype( 'float' )                                    \
                    .to_frame( 'var' )                                     \
                    .reset_index( drop = True )
                  ], axis = 1
                )

Примечание. Чтобы получить минимальный пример, его можно легко установить на подмножество (например, с помощью df.loc[df['time'] <= 400, :]), но, поскольку я все равно имитирую данные, я подумал, что исходный размер даст лучший обзор.

Для каждой определенной группы ['measurement_id', 'time', 'group']мне нужно вызвать следующую функцию:

from sklearn.cluster import SpectralClustering
from pandarallel     import pandarallel

def cluster( x, index ):
    if len( x ) >= 2:
        data = np.asarray( x )[:, np.newaxis]
        clustering = SpectralClustering( n_clusters   =  5,
                                         random_state = 42
                                         ).fit( data )
        return pd.Series( clustering.labels_ + 1, index = index )
    else:
        return pd.Series( np.nan, index = index )

Для повышения производительности я попробовал два подхода:

Pandarallel пакет

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

pandarallel.initialize( progress_bar = True )
df \
  .groupby( ['measurement_id', 'time', 'group'] ) \
  .parallel_apply( lambda x: cluster( x['var'], x['object'] ) )

Тем не менее, это кажется неоптимальным, поскольку он потребляет много оперативной памяти, и не все ядра используются в вычислениях (даже несмотря на то, что в pandarallel.initialize()методе явно указано количество ядер ). Кроме того, иногда вычисления завершаются с различными ошибками, хотя у меня не было возможности найти причину этого (возможно, нехватка оперативной памяти?).

PySpark Pandas UDF

Я также дал UDF Spark Pandas, хотя я совершенно новичок в Spark. Вот моя попытка:

import findspark;  findspark.init()

from pyspark.sql           import SparkSession
from pyspark.conf          import SparkConf
from pyspark.sql.functions import pandas_udf, PandasUDFType
from pyspark.sql.types     import *

spark = SparkSession.builder.master( "local" ).appName( "test" ).config( conf = SparkConf() ).getOrCreate()
df = spark.createDataFrame( df )

@pandas_udf( StructType( [StructField( 'id', IntegerType(), True )] ), functionType = PandasUDFType.GROUPED_MAP )
def cluster( df ):
    if len( df['var'] ) >= 2:
        data = np.asarray( df['var'] )[:, np.newaxis]
        clustering = SpectralClustering( n_clusters   =  5,
                                         random_state = 42
                                         ).fit( data )
        return pd.DataFrame( clustering.labels_ + 1,
                             index = df['object']
                             )
    else:
        return pd.DataFrame( np.nan,
                             index = df['object']
                             )

res = df                                           \
        .groupBy( ['id_half', 'frame', 'team_id'] ) \
        .apply( cluster )                            \
        .toPandas()

К сожалению, производительность также была неудовлетворительной, и из того, что я читал по этой теме, это может быть просто бременем использования функции UDF, написанной на Python, и связанной с этим необходимости преобразования всех объектов Python в объекты Spark и обратно.

Итак, вот мои вопросы:

  1. Можно ли отрегулировать любой из моих подходов, чтобы устранить возможные узкие места и повысить производительность? (например, настройка PySpark, настройка неоптимальных операций и т. д.)
  2. Есть ли у них лучшие альтернативы? Как они соотносятся с предоставленными решениями с точки зрения производительности?
Kuba_
источник
2
ты исследовал дневник ?
Данила Гончар
1
Пока нет, но спасибо за ваше предложение - я
попробую
к сожалению я не работал с dask(((так что мой комментарий это просто совет для исследования.
Данила Ганчар
Под производительностью я подразумевал время, за которое можно завершить вычисления.
Куба_

Ответы:

1

В : « Можно ли отрегулировать любой из моих подходов, чтобы устранить возможные узкие места и повысить производительность? (Например, настройка PySpark, настройка неоптимальных операций и т. Д.) »

+1за упоминание установки надстройки накладных расходов для каждой стратегии вычисления. Это всегда создает точку безубыточности, только после этого нестратегия [SERIAL]может достичь какой-либо полезной радости от некоторого [TIME]ускорения желаемого для -домена домена (но, если иное, как правило, [SPACE]-Дополнения стоимости домена позволяют или остаются выполнимыми - да, ОЗУ). .. наличие и доступ к устройству такого размера, бюджету и другим подобным ограничениям реального мира)

Во-первых,
проверка перед полетом, перед тем как мы взлетим

. Новая, неукоснительная формулировка Закона Амдала в настоящее время может включать обе эти дополнительные pSO + pTOнакладные расходы и отражает их при прогнозировании достижимых уровней ускорения, включая безубыточность. точка, с которой может стать значимым (в смысле затраты / эффект, эффективность) идти параллельно.

введите описание изображения здесь

Тем не менее,
это не наша основная проблема здесь .
Это идет дальше:

Далее,
учитывая вычислительные затраты SpectralClustering(), которые здесь будут использоваться для использования ядра радиальной функции Больцмана ~ exp( -gamma * distance( data, data )**2 ), по-видимому, нет никакого dataпреимущества от разделения -объекта на любое количество несвязанных рабочих единиц, так как distance( data, data )-компонент по определению должен посетить все data-элементы (см. затраты на коммуникацию для { process | node }топологий с произвольной передачей значений по очевидным причинам ужасно плохи, если не наихудшие сценарии использования для { process | node }-распределенной обработки, если не прямые анти-шаблоны (за исключением некоторых действительно загадочных, не требующих памяти / не имеющих состояния, но все же вычислительных структур).

Для педантичных аналитиков, да - прибавьте к этому (и мы уже можем сказать, плохое состояние) затраты - опять-таки - любой-к-любому- k-средних - обработки , здесь, O( N^( 1 + 5 * 5 ) )что об этом идет N ~ len( data ) ~ 1.12E6+, ужасно против нашего желания иметь некоторые умная и быстрая обработка.

И что?

Несмотря на то, что затраты на настройку не игнорируются, повышенные расходы на связь почти наверняка отключат любое улучшение от использования описанных выше попыток перехода от чисто [SERIAL]технологического потока в некую форму просто - [CONCURRENT]или истинной [PARALLEL]оркестровки некоторых рабочих подразделений. , из-за увеличения накладных расходов, связанных с необходимостью реализации (тандемной пары) топологий передачи любого значения любому.

Если бы не они?

Что ж, это звучит как оксюморон вычислительной науки - даже если бы это было возможно, стоимость предварительно вычисленных расстояний «от любого к любому» (что потребовало бы этих огромных [TIME]затрат) - «заранее» - сложность домена (где? Как? другие, неустранимые задержки, допускающие возможную маскировку задержек с помощью некоторого (пока неизвестного) постепенного наращивания полной в будущем матрицы расстояния «любой-к-любому»?)), но переместили бы эти принципиально существующие затраты в какое-то другое место в [TIME]- и [SPACE]-Домены, а не сокращать их.

Q : "Есть ли у них лучшие альтернативы? "

Единственное, что я знаю до сих пор, - это попытаться, если проблема может быть переформулирована в другую, сформулированную в QUBO, проблемную моду (см .: Q- qantum- U nconstrained- B inary- O ptimisation). хорошая новость заключается в том, что инструменты для этого существуют, база знаний из первых рук и практический опыт решения проблем существуют и расширяются)

Q : Как они сравниваются с предоставленными решениями с точки зрения производительности?

Производительность захватывает дух - проблема, сформулированная O(1)в [TIME]QUBO, имеет многообещающий (!) Решатель в постоянном времени (в -Domain) и несколько ограничен в [SPACE]-Domain (где недавно анонсированные трюки LLNL могут помочь избежать этого физического мира, текущей реализации QPU, ограничения проблемы размеры).

user3666197
источник
Это интересный ответ, но, похоже, он упускает суть - ОП обучает несколько маленьких моделей, а не одну. Таким образом, ваше основное наблюдение в основном не имеет значения.
user10938362
@ user10938362 Как ваша заявленная собственность (обучающие маленькие модели) переходит в другую, не указанную выше метрику затрат на обработку, указанную выше? Конечно, многие меньшие модели обещают теоретически просто линейно растущую сумму (все еще) больших затрат на индивидуальную (теперь меньшую по N, но не по другим факторам) обработку, но вы должны добавить к этому значительно более дорогую сумму всех дополнительные расходы, связанные как с накладными расходами на установку, так и с завершением, плюс все дополнительные накладные расходы на связь (параметры / данные / результаты + обычно также пары затрат на обработку SER / DES на каждом шаге)
user3666197
0

Это не ответ, но ...

Если вы бежите

df.groupby(['measurement_id', 'time', 'group']).apply(
    lambda x: cluster(x['var'], x['object']))

(т. е. только с пандами), вы заметите, что вы уже используете несколько ядер. Это потому, что sklearnиспользует joblibпо умолчанию для распараллеливания работы. Вы можете поменять планировщик в пользу Dask и, возможно, получить большую эффективность за счет совместного использования данных между потоками, но до тех пор, пока работа, которую вы выполняете, связана с процессором, как это, вы ничего не сможете сделать, чтобы ускорить его.

Короче говоря, это проблема алгоритма: выясните, что вам действительно нужно вычислить, прежде чем пытаться рассмотреть различные основы для его вычисления.

mdurant
источник
Не могли бы вы объяснить, почему вы упоминаете «… разделение данных между потоками …» после того, как разделение работы было организовано joblibпорожденными процессами , которые не имеют ничего общего с потоками, тем более с разделением? Спасибо за ваше любезное разъяснение аргументов.
user3666197
Точно, jboblib обычно использует процессы, но в качестве альтернативы он может использовать dask в качестве бэкэнда, где вы можете выбрать смесь потоков и процессов.
Мдурант
Я новичок в параллельных вычислениях, но даже если sklearn использует параллелизацию, разве это не бесполезно в этих настройках? Я имею в виду, что операции, выполняемые sklearn, чрезвычайно просты, поскольку каждая операция кластеризации применяется только к 10 точкам. Опять же, я могу ошибаться, но я думаю, что реальная проблема параллелизации обработки фрагментов исходных данных.
Куба_
«
Разве
0

Я не являюсь экспертом Dask, но в качестве основы я предоставляю следующий код:

import dask.dataframe as ddf

df = ddf.from_pandas(df, npartitions=4) # My PC has 4 cores

task = df.groupby(["measurement_id", "time", "group"]).apply(
    lambda x: cluster(x["var"], x["object"]),
    meta=pd.Series(np.nan, index=pd.Series([0, 1, 1, 1])),
)

res = task.compute()
чокнутый
источник