Okay, I take it back, split_every=1 as a default for shuffling is a bad idea. But I still think it's worthwhile to have it as an option, as I do think it is the right choice when there are few, large partitions per worker.

This is largely orthogonal to the choice of when to default to the shuffle-based groupby, which I still believe should just be when split_out > 1.

Thanks for taking a look @rjzamora! I'm now backing off of my initial proposal to make split_every default to 1, but I think I'd still advocate for it to be allowed to be one in this case.

I don't like that I haven't been able to think of a good heuristic for what it should be, but all I can think of involves comparing number of partitions, representative partition sizes, and number of workers.

Thanks for taking a look @rjzamora! I'm now backing off of my initial proposal to make split_every default to 1, but I think I'd still advocate for it to be allowed to be one in this case.

That seems fair to me. For both the shuffle and aca algorithms, split_every is meant to specify how many chunk-task outputs can be safely (and perhaps efficiently) concatenated together. In the aca algo, this is used to specify the "k" in the k-ary reduction tree. While in the shuffle algo, this tells us how many adjacent partitions should be coelesced after the initial blockwise chunk aggregation. These concepts are very similar, but the lower limits happen to slightly different.

Take-away: We should indeed support a distinct "lower-limit" on split_every for the shuffle-based algorithm.

it looks like there's good agreement -- is this safe to merge?

I think the only sticking point for me is that I'd like to avoid using shuffle="disk" by default on the threaded scheduler - I'd expect that to be a significant performance regression compared to shuffle=False. Therefore, we need to use shuffle="tasks" explicitly, or somehow make the shuffle default scheduler-dependent.

Disk is much better for the local scheduler. It has the same scalability
and out of memory properties as the p2p shuffle

This is probably true for larger-than-memory data, but I definitely see a significant performace regression for "large" data (5-6GB) that still fits comfortably in memory:

from dask.datasets import timeseries

ddf = timeseries(end='2002-01-01', id_lam=1e12)
%time ddf.groupby("id").agg({"x": "mean"}, split_out=4, shuffle=False).compute()
# Wall time: 12.9 s

%time ddf.groupby("id").agg({"x": "mean"}, split_out=4, shuffle="tasks").compute()
# Wall time: 13.2 s

%time ddf.groupby("id").agg({"x": "mean"}, split_out=4, shuffle="disk").compute()
# Wall time: 31 s

Generally speaking, the shuffle-vs-aca decision is a bit more complicated for the threaded scheduler.

From my end, this is ready. Is there anything else to be done from your perspective @rjzamora?