TopolearnShuffles are the most expensive thing a distributed DataFrame does, and choosing the right join strategy can change runtimes by orders of magnitude.
Many-partition data has to be shuffled whenever rows with the same key must end up in the same partition: merges on non-index columns, set_index, sort_values, groupby().apply, drop_duplicates with split_out. A shuffle moves almost every row across the network, and its cost grows with data size.
Dask chooses a merge strategy based on what it knows:
- Broadcast join: if one side is small (or has one partition), it is sent to every partition of the big side. No shuffle. You can force it with
broadcast=Trueor byrepartition(npartitions=1)on the small table. - Index-aligned join: if both sides are indexed with known, compatible divisions, partitions are joined pairwise. No shuffle.
- Hash shuffle join: otherwise both sides are partitioned by hashing the key, then joined partition by partition.
On a distributed cluster the default shuffle method is P2P (peer-to-peer), which streams data between workers with bounded memory and spills to disk. It is far more robust than the older task-based shuffle, which creates a number of tasks quadratic in the partitions.
Tactics: filter and select columns before merging, merge small dimension tables by broadcast, pre-partition data by a frequently joined key and save it with divisions, and avoid sort_values unless you really need global order.
Example
import dask.dataframe as dd
import pandas as pd
facts = dd.from_pandas(pd.DataFrame({
"uid": [1, 2, 3, 1, 2, 3] * 5, "amount": range(30)}), npartitions=5)
users = pd.DataFrame({"uid": [1, 2, 3], "country": ["NO", "SE", "NO"]})
# Small pandas side: Dask broadcasts it to every partition (no shuffle)
joined = facts.merge(dd.from_pandas(users, npartitions=1), on="uid", how="left")
print(joined.npartitions) # 5
print(joined.groupby("country").amount.sum().compute().sort_index().to_dict())
# {'NO': 290, 'SE': 145}
# Two large sides on a non-index key: hash shuffle
other = dd.from_pandas(pd.DataFrame({"uid": [1, 2, 3] * 4, "score": range(12)}),
npartitions=4)
big = facts.merge(other, on="uid", shuffle_method="tasks")
print(len(big)) # 120Remember
- Merges on non-index keys,
set_indexand sorts trigger shuffles - Broadcast small tables; align on known divisions to avoid shuffling
- The P2P shuffle on distributed is the robust default
Watch out for
- Merging two huge tables before filtering them, shuffling data you'll throw away.