Groupby in parallel

[bicepjai]

I had a questions with regards to how to perform groupby in parallel.
can you provide sample notebooks on this task ?
https://discourse.julialang.org/t/dataframes-with-groupby-in-parallel/47186

[bkamins]

Can you please paste here a working (and reproducible so that I can run it) example of what you want that runs on a single thread? Thank you!

(in general DataFrames.jl natively does not support multi-threading yet, but maybe in your use case it will be possible to propose something)

[bicepjai]

Dataframe is of size 2450225 rows and 115 columns.
The code for the simple groupby and combine each sub dataframe

sortbydates = Dates.Date(1980,1,1):Dates.Day(3):Dates.Date(2019,1,1)
groupbydates = Dates.Date(1980,1,1):Dates.Month(3):Dates.Date(2019,1,1)

df = DataFrame(
        ticker = rand(1:15000,2450225), 
        groupbycol = rand(groupbydates,2450225),
        sortbycol = rand(sortbydates,2450225), 
)

for i in 1:100
    insertcols!(df,
        i+3,
        "column"*string(1+i) => 1:2450225,
        makeunique=true)
end
size(df)

@time df1 = @pipe df |>
        groupby(_, [:ticker, :groupbycol]) |>
        combine(_) do df_
            first(sort(df_, :sortbycol), 1)
        end
df1

The above mocked up date takes around 2 minutes.

---

The content of my dataframe can be seen here

When the groupbycol is of type Date. The process takes 12 minutes

@time data_sf1_1 = @pipe data_sf1 |>
        groupby(_, [:ticker, :calendardate]) |>
        combine(_) do df_
            first(sort(df_, :datekey), 1)
        end
data_sf1_1

output

709.155233 seconds (246.36 M allocations: 1.048 TiB, 12.05% gc time)
576,564 rows × 112 columns (omitted printing of 95 columns)

When the groupbycol changed to categorical string type variable. The process takes 1.5 minutes

@time data_sf1_1 = @pipe data_sf1 |>
        groupby(_, [:ticker, :quarter]) |>
        combine(_) do df_
            first(sort(df_, :datekey), 1)
        end
data_sf1_1

output

 77.023839 seconds (23.97 M allocations: 105.898 GiB, 12.85% gc time)
55,639 rows × 112 columns (omitted printing of 94 columns)

I have lot of cores and ram to throw at, looking for a faster way
For reference. I am a a big pandas user and in the process to change my workflow to
Julia to take advantage of its features.
The pandas groupby in parallel using multiprocessing module executes in around 35 seconds

[bkamins]

First, if you do this:

@time df1 = @pipe df |>
        groupby(_, [:ticker, :groupbycol]) |>
        combine(_) do df_
            df_[argmin(df_.sortbycol), :]
        end

it should be much faster (I think I have already mentioned it somewhere)

If you still want to be faster (we are getting closer to multi-threading, but we still use a single thread):

@time begin
    gdf = groupby(df, [:ticker, :groupbycol])
    df[[parentindices(sdf)[1][argmin(sdf.sortbycol)] for sdf in gdf], :]
end

It should now be much faster than Pandas and it is fast enough that I do not think multi-threading will help much here. On my laptop it takes ~3 seconds.

Now let us go to your original code. You can parallelize it by:

@time begin
    gdf = groupby(df, [:ticker, :groupbycol])
    dfrs = Vector{Any}(undef, length(gdf))
    @time Threads.@threads for i in 1:length(gdf)
        dfrs[i] = first(sort(gdf[i], :sortbycol))
    end
    DataFrame(dfrs)
end

but note that this is super wasteful (thus slow), as with each sort(gdf[i], :sortbycol) call you are allocating a small DataFrame that has to be stored and passed around (so you are probably likely to hit GC problems unless you really have a lot of RAM).

[bicepjai]

First, if you do this:

@time df1 = @pipe df |>
        groupby(_, [:ticker, :groupbycol]) |>
        combine(_) do df_
            df_[argmin(df_.sortbycol), :]
        end

On my dataset, it took 12 seconds for the above operation

If you still want to be faster (we are getting closer to multi-threading, but we still use a single thread):

@time begin
    gdf = groupby(df, [:ticker, :groupbycol])
    df[[parentindices(sdf)[1][argmin(sdf.sortbycol)] for sdf in gdf], :]
end

On my dataset, it took less than 1 second for the above operation

Thanks for the sample codes.
Why is the second method much faster than the first one ?
Is it because the df_ in first method gets allocated for each combine
where as the second one the allocation is never done ?

Also the second one is hard to decipher and would take more time to write
where as the first one seems very natural

[bkamins]

Yes - the first method allocates more. The second method uses low-level API (but exposed - I do not use anything internal), which probably even most users are not aware about. It allocates but only the vector of indices in the parent data frame (which is fast, because it is just a vector of integers as opposed to some more complex object). Also the second method does not call all the machinery that combine has internally (it potentially does many more things - check the docsting - and handling this must cost).

In summary, as usual in Julia:

1. the less you allocate the faster your code can run
2. you can get a very fast code if you want to specialize it for your use case (and I would assume this would be hard with Python)
3. (this is a general feature of everything I guess) general purpose functions will be slower than custom code for a specific use-case

[bicepjai]

Thanks for the explanation.

[bicepjai]

@bkamins when we have complicated combine other than the argmin(sdf.sortbycol) as used in this example; this parallel processing again becomes an issue. any elegant way to solve complicated combine ?

[bkamins]

Can you please post a new issue with an exact example of what you are struggling with? Here and here you have a description where and how multi-threading is currently supported in DataFrames.jl.