MultiProcessingBenchmark/MultiProc.py at master · srivassid/MultiProcessingBenchmark · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
import pandas as pd
import numpy as np
import time
from multiprocessing import Pool
from .CreateDataframeClass import CreateDataFrame
from .SimpleStatisticsClass import MonoSimpleStatistics, MonoUtilityFunctions, GroupByAggregateLoop, GroupByAggregateNoLoop
from .MultiProcClass import MultiSimpleStatistics, MultiUtilityFunctions, MultiGroupByAggregateNoLoop, MultiGroupByAggregateLoop
from .CreatePlottingClass import PlotTimes

pd.options.display.width = 0
pd.set_option('display.float_format', lambda x: '%.3f' % x)

class MultiProcessingClass():

    def __init__(self):
        pass

    def MultiSimpleStatistics(self, df, n_cores):
        '''Call Simple Statistics functions'''
        self.multi_proc = MultiSimpleStatistics()

        self.simple_stats = MonoSimpleStatistics()

        # mean
        self.multi_proc.mean_parallelize_dataframe(df, self.simple_stats.mean_stat, n_cores)

        # sum
        self.multi_proc.sum_parallelize_dataframe(df, self.simple_stats.sum_stat, n_cores)

        # count
        self.multi_proc.count_parallelize_dataframe(df, self.simple_stats.count_stat, n_cores)

        # stdev time
        self.multi_proc.std_parallelize_dataframe(df, self.simple_stats.stdev_stat, n_cores)

        # rolling mean time
        self.multi_proc.roll_parallelize_dataframe(df, self.simple_stats.roll_mean_stat, n_cores)

        self.multi_proc.to_csv()

    def multiUtilityFunctions(self, df, df_other, val, n_cores):
        '''Call multi utility functions'''
        self.multi_util = MultiUtilityFunctions()
        self.mono_util = MonoUtilityFunctions()

        self.multi_util.sorting(df.sample(frac=1), self.mono_util.sorting, n_cores)
        self.multi_util.searching(df.apply(lambda x:round(x,2)), self.mono_util.searching, val, n_cores)
        self.multi_util.merging(df,df_other,self.mono_util.merging, n_cores)
        self.multi_util.merging_asof(df,df_other,self.mono_util.merge_asof, n_cores)
        self.multi_util.join(df,df_other,self.mono_util.join, n_cores)
        self.multi_util.concat(df,df_other,self.mono_util.concat, n_cores)
        self.multi_util.to_csv()

    def multiAggregateNoLoop(self, df, n_cores):
        """Call aggregate functions with multi cores no loop"""
        self.multiAgg = MultiGroupByAggregateNoLoop()
        self.monoAgg = GroupByAggregateNoLoop()
        self.multiAgg.groupByAggregateNoLoop(df, self.monoAgg.groupByAggregateNoLoop, n_cores)

    def multiAggregateLoop(self, df, n_cores):
        """Call aggregate functions with multi cores no loop"""
        self.multiAgg = MultiGroupByAggregateLoop()
        self.monoAgg = GroupByAggregateLoop()
        self.multiAgg.groupByAggregateLoop(df, self.monoAgg.groupByAggregateLoop, n_cores)