洗牌性能
目录
洗牌性能¶
诸如 groupby、join 和 set_index 之类的操作，由于 Dask DataFrame 的并行、大于内存和分布式特性，具有与普通 Pandas 不同的特殊性能考虑。
简单情况¶
首先，对于已知归约（如 mean, sum, std, var, count, nunique），常见的 groupby 操作（如 df.groupby(columns).reduction()）都非常快速高效，即使分区没有按已知划分方式清晰划分也是如此。这是常见的情况。
此外，如果划分方式已知，并且分组列包含索引，则将任意函数应用于分组是高效的。
将 Dask DataFrame 连接到 Pandas DataFrame，或者沿索引连接两个 Dask DataFrame 时，连接操作也相当快。在这些常见情况下操作时，无需进行特殊考虑。
因此，如果您正在进行常见的 groupby 和 join 操作，那么您可以停止阅读本文了。一切都将很好地扩展。幸运的是，大多数情况下都是如此
>>> ddf.groupby(columns).known_reduction() # Fast and common case
>>> ddf.groupby(columns_with_index).apply(user_fn) # Fast and common case
>>> ddf.join(pandas_df, on=column) # Fast and common case
>>> lhs.join(rhs) # Fast and common case
>>> lhs.merge(rhs, on=columns_with_index) # Fast and common case
复杂情况¶
在某些情况下，例如将任意函数应用于分组（当不是按已知划分方式对索引分组时）、沿非索引列进行连接，或显式将未排序的列设置为索引时，我们可能需要触发完整数据集洗牌。
>>> ddf.groupby(columns_no_index).apply(user_fn) # Requires shuffle
>>> lhs.join(rhs, on=columns_no_index) # Requires shuffle
>>> ddf.set_index(column) # Requires shuffle
当我们需要沿新索引重新排序数据时，洗牌是必需的。例如，如果我们有按时间组织的银行记录，现在想按用户 ID 组织它们，那么我们就需要移动大量数据。在 Pandas 中，所有这些数据都适合内存，因此此操作很容易。现在我们不再假设所有数据都适合内存，必须更加小心一些。
通过将操作限制在上面提到的简单情况，可以避免重新排序数据。
洗牌方法¶
当前有两种数据洗牌策略，取决于您是在单台机器上还是在分布式集群上：磁盘洗牌和网络洗牌。
网络洗牌¶
在分布式集群上操作时，Dask worker 可能无法访问共享硬盘。在这种情况下，我们通过根据输入分区最终到达的位置将其分成许多小块，并在网络中移动这些小块来洗牌数据。
选择方法¶
Dask 默认使用磁盘洗牌，但如果默认调度器设置为使用 dask.distributed.Client，例如用户将 Client 设置为默认值时，则会切换到分布式洗牌算法。
client = Client('scheduler:8786', set_as_default=True)
或者，如果您希望避免默认设置，可以使用 dataframe.shuffle.method 配置选项配置全局洗牌方法。这可以全局设置。
dask.config.set({"dataframe.shuffle.method": "p2p"})
ddf.groupby(...).apply(...)
或作为上下文管理器。
with dask.config.set({"dataframe.shuffle.method": "p2p"}):
ddf.groupby(...).apply(...)
此外，set_index 也接受一个 shuffle_method 关键字参数，可用于选择磁盘洗牌或基于任务的洗牌。
ddf.set_index(column, shuffle_method='disk')
ddf.set_index(column, shuffle_method='tasks')
ddf.set_index(column, shuffle_method='p2p')
聚合¶
Dask 支持 Pandas 的 aggregate 语法，以便在同一组上运行多个归约。直接支持常见的归约，例如 max ， sum ， list 和 mean。
>>> ddf.groupby(columns).aggregate(['sum', 'mean', 'max', 'min', list])
Dask 还支持用户自定义归约。为了确保良好的性能，归约必须分解为三个独立的步骤。chunk 步骤独立应用于每个分区，并归约分区内的数据。aggregate 将分区内的结果合并。可选的 finalize 步骤合并从 aggregate 步骤返回的结果，并应返回一个最终的列。为了让 Dask 识别该归约，它必须作为 dask.dataframe.Aggregation 的实例传入。
例如，sum 可以实现为：
custom_sum = dd.Aggregation('custom_sum', lambda s: s.sum(), lambda s0: s0.sum())
ddf.groupby('g').agg(custom_sum)
名称参数应不同于现有归约，以避免数据损坏。每个函数的参数都是预分组的 Series 对象，类似于 df.groupby('g')['value']。
许多归约只能使用多个临时变量来实现。为了实现这些归约，步骤应该返回元组并期望多个参数。均值函数可以实现为：
custom_mean = dd.Aggregation(
'custom_mean',
lambda s: (s.count(), s.sum()),
lambda count, sum: (count.sum(), sum.sum()),
lambda count, sum: sum / count,
)
ddf.groupby('g').agg(custom_mean)
例如，让我们计算 DataFrame 的按组范围（最大值 - 最小值）。
>>> df = pd.DataFrame({
... 'a': ['a', 'b', 'a', 'a', 'b'],
... 'b': [0, 1, 0, 2, 5],
... })
>>> ddf = dd.from_pandas(df, 2)
我们定义了构建块，用于查找每个块的最大值和最小值，然后是所有块的最大值和最小值。最后，我们通过计算包含最大值和最小值的 Series 之间的差值来完成。
>>> def chunk(grouped):
... return grouped.max(), grouped.min()
>>> def agg(chunk_maxes, chunk_mins):
... return chunk_maxes.max(), chunk_mins.min()
>>> def finalize(maxima, minima):
... return maxima - minima
最后，我们创建并使用该聚合。
>>> extent = dd.Aggregation('extent', chunk, agg, finalize=finalize)
>>> ddf.groupby('a').agg(extent).compute()
b
a
a 2
b 4
要将 dask.dataframe.groupby.SeriesGroupBy.nunique 应用于多列，您可以使用：
>>> df['c'] = [1, 2, 1, 1, 2]
>>> ddf = dd.from_pandas(df, 2)
>>> nunique = dd.Aggregation(
... name="nunique",
... chunk=lambda s: s.apply(lambda x: list(set(x))),
... agg=lambda s0: s0.obj.groupby(level=list(range(s0.obj.index.nlevels))).sum(),
... finalize=lambda s1: s1.apply(lambda final: len(set(final))),
... )
>>> ddf.groupby('a').agg({'b':nunique, 'c':nunique})
要访问 NumPy 函数，请使用带有 lambda 函数的 apply，例如 .apply(lambda r: np.sum(r))。以下是平方和聚合的一个示例：
>>> dd.Aggregation(name="sum_of_squares", chunk=lambda s: s.apply(lambda r: np.sum(np.power(r, 2))), agg=lambda s: s.sum())