Groupby dictionary aggregation regression

[amyskov]

System information

• OS Platform and Distribution (e.g., Linux Ubuntu 16.04): Ubuntu 20.04
• Modin version : aa818f5
• Python version: 3.8.6
• Code we can use to reproduce:

import os

os.environ["MODIN_ENGINE"] = "ray"
os.environ["MODIN_CPUS"] = "4"

import modin.pandas as pd
from timeit import default_timer as timer

data_filename = "h2o_10k.csv"

df = pd.read_csv(data_filename)


def q1(df):
    groupby_cols = ["id1"]
    agg_cols_funcs = {"v1": "sum"}

    return df.groupby(groupby_cols).agg(agg_cols_funcs)

def q10(df):
    groupby_cols = ["id1", "id2", "id3", "id4", "id5", "id6"]
    agg_cols_funcs = {"v3": "sum", "v1": "count"}

    return df.groupby(groupby_cols).agg(agg_cols_funcs)

def meas_func(func, iters, *args, **kwargs):
    t_meas = float("inf")
    for _ in range(iters):
        t0 = timer()
        ans = func(*args, **kwargs)
        ans.shape
        t = timer() - t0
        del ans
        t_meas = t if t < t_meas else t_meas
    
    return t_meas

t_q1 = meas_func(q1, iters=3, **{"df": df})
t_q10 = meas_func(q10, iters=3, **{"df": df})

# results comparison between before vs after ee39d1767efb363fe6c147fc701b41315eee3afa commit
print("t_q1:", t_q1) # 0.026 s vs 0.031s
print("t_q10:", t_q10) # 0.091 s vs 2.09 s

Describe the problem

Regression occurred after ee39d17 commit.
data file h2o_10k.txt

Source code / logs

[dchigarev]

The reason for the slow dict-groupby is pretty ridiculous. There is a dict, that contains map and reduce functions for every operation that supports GroupbyReduce. These functions are applied to groups in every partition.

modin/modin/data_management/functions/groupby_function.py

Lines 234 to 267 in aa818f5

	groupby_reduce_functions = {
	"all": (
	lambda df, *args, **kwargs: df.all(*args, **kwargs),
	lambda df, *args, **kwargs: df.all(*args, **kwargs),
	),
	"any": (
	lambda df, *args, **kwargs: df.any(*args, **kwargs),
	lambda df, *args, **kwargs: df.any(*args, **kwargs),
	),
	"count": (
	lambda df, *args, **kwargs: df.count(*args, **kwargs),
	lambda df, *args, **kwargs: df.sum(*args, **kwargs),
	),
	"max": (
	lambda df, *args, **kwargs: df.max(*args, **kwargs),
	lambda df, *args, **kwargs: df.max(*args, **kwargs),
	),
	"min": (
	lambda df, *args, **kwargs: df.min(*args, **kwargs),
	lambda df, *args, **kwargs: df.min(*args, **kwargs),
	),
	"prod": (
	lambda df, *args, **kwargs: df.prod(*args, **kwargs),
	lambda df, *args, **kwargs: df.prod(*args, **kwargs),
	),
	"size": (
	lambda df, *args, **kwargs: pandas.DataFrame(df.size(*args, **kwargs)),
	lambda df, *args, **kwargs: df.sum(*args, **kwargs),
	),
	"sum": (
	lambda df, *args, **kwargs: df.sum(*args, **kwargs),
	lambda df, *args, **kwargs: df.sum(*args, **kwargs),
	),
	}

The problem is that they're all lambdas, for pandas all of them are unknown python callables, so it can't execute aggregation with its fast C module, we force it to apply our callables. If we replace callables to the function names it performs much faster!

Here is a comparison of groupby aggregation in pandas with function name and with callable:

import pandas
from timeit import default_timer as timer

# 500mb H2O dataset
data_filename = "/localdisk/benchmark_datasets/h2o/G1_1e7_1e1_0_0.csv"
df = pandas.read_csv(data_filename)

t1 = timer()
df.groupby("id6").agg({"v1": "sum"})
print(f"str function: {timer() - t1}")

t1 = timer()
df.groupby("id6").agg({"v1": lambda df: df.sum()})
print(f"callable function: {timer() - t1}")

# Output:
# str function: 0.7193635171279311
# callable function: 38.2026267433539

Speaking about modin, when I replaced lambdas to function names in groupby_reduce_functions and ran the script from the issue header I got this results:

Before ee39d17:
t_q1: 0.025292523205280304
t_q10: 0.08418524172157049

After ee39d17:
t_q1: 0.030419232323765755
t_q10: 1.9959391057491302

After ee39d17 + fix(lambdas -> str-names):
t_q1: 0.023051242344081402
t_q10: 0.0943737206980586

Going to create PR that fixes this soon

[dchigarev]

results of H2O queries with 500mb dataset and 112 partitions:

Before ee39d17:
t_q1: 3.1282251700758934
t_q10: 144.17810649238527

After ee39d17 + fix(lambdas -> str-names):
t_q1: 0.3956367587670684
t_q10: 140.4794331882149