在 Dask 中实施等宽间隔特征工程

Question

在等宽离散化中，变量值被分配到相同宽度的区间。间隔的数量是用户定义的，宽度由最小/最大值和间隔的数量确定。

例如，给定值 10、20、100、130，最小值为 10，最大值为 130。如果用户将间隔数定义为 6，则给出以下公式：

区间宽度 = (Max(x) - Min(x)) / N

宽度为 (130 - 10) / 6 = 20

六个从零开始的区间是：[ 10, 30, 50, 70, 90, 110, 130]

最后，为数据集中的每个元素定义区间分配：

Value in the dataset    New feature engineered value
          10                      0
          20                      0
          57                      2 
         101                      4
         130                      5

我有以下代码，它使用pandas数据框和sklean函数将数据框以等宽间隔划分：

from sklearn.preprocessing import KBinsDiscretizer
discretizer = KBinsDiscretizer(n_bins=10, encode='ordinal', strategy='uniform')
df['output_col'] = discretizer.fit_transform(df[['input_col']])

这工作正常，但我需要实现一个等效的daskKBinsDiscretizer函数，该函数将在多个分区中并行触发该进程，并且我在dask_ml.preprocessing任何建议中找不到？我无法使用，map_partitions因为它将将该函数独立地应用于每个分区，并且我需要将间隔应用于整个数据帧。

Answer 1

您面临着分布式工作流程的常见权衡。您是否想花费时间/资源/计算来确定确切的最小值/最大值（这是您描述的分箱方案的先决条件），或者是一个近似答案吗？如果是后者，您如何设计一种算法，在保持高效的同时充分捕获数据的最小值/最大值？

我们可以从精确的解决方案开始，因为它更容易实现。关键是先找到最小值和最大值，然后将数据数字化。请注意，这需要计算列中的所有值两次。如果可以选择保留数据（例如，您正在使用分布式集群或可以将列放入内存中），这将有助于避免不必要的重复：

def discretize_exact(
    s: dask.dataframe.Series, K: int
) -> dask.dataframe.Series:
    """
    Discretize values in dask.dataframe Series into K equal-width bins

    Parameters
    ----------
    s : dask.dataframe.Series
        Series with values to be binned
    K : int
        Number of equal-width bins to generate

    Returns
    -------
    binned : dask.dataframe.Series
        dask.dataframe.Series with scheduled np.digitize operation
        called using map_partitions. The values in ``binned`` will
        be in [0, K] giving the index of the K bins in the interval
        [vmin, vmax].
    """

    # schedule the min/max computation
    vmin, vmax = s.min(), s.max()

    # compute vmin and vmax together so we only compute once
    vmin, vmax = dask.compute(vmin, vmax)

    # will create K - 1 equal width bins, with
    # the outer ends open, such that the first bin will be
    # (-inf, vmin + step) and the last will be [vmax - step, inf)
    bins = np.linspace(vmin, vmax, (K + 1))[1:-1]

    return s.map_partitions(
        np.digitize,
        bins=bins,
        meta=('binned', 'uint16'),
    )

这确实（我认为）您正在寻找的东西，但确实涉及在安排装箱操作之前首先计算最小值和最大值。使用示例框架：

import dask.dataframe, pandas as pd, numpy as np
N = 10000
df = dask.dataframe.from_pandas(
    pd.DataFrame({'a': np.random.random(size=N)}),
    chunksize=1000,
)

我们可以使用上面的函数来离散化我们的数据：

In [68]: df['binned_a'] = discretize_exact(df['a'], K=10)

In [69]: df
Out[69]:
Dask DataFrame Structure:
                      a binned_a
npartitions=10
0               float64   uint16
1000                ...      ...
...                 ...      ...
9000                ...      ...
9999                ...      ...
Dask Name: assign, 40 tasks

In [70]: df.compute()
Out[70]:
             a  binned_a
0     0.548415         5
1     0.872668         8
2     0.466869         4
3     0.133986         1
4     0.833126         8
...        ...       ...
9995  0.223438         2
9996  0.575271         5
9997  0.922593         9
9998  0.030127         0
9999  0.204283         2

[10000 rows x 2 columns]

或者，您可以尝试近似垃圾箱边缘。您可以通过多种方式执行此操作，包括对数据帧进行采样以识别一个或多个分区的最小值/最大值，或者用户可以提供过于宽泛的范围估计。请注意，根据您的工作流程，计算第一个分区可能仍涉及计算整个图的很大一部分，甚至是整个图（如果例如数据帧在最近的步骤中被重新洗牌）。

def find_minmax_of_first_partition(
    s: dask.dataframe.Series
) -> tuple[float, float]:
    """
    Find the min and max of the first partition of a dask.dataframe.Series
    """

    partition_0_stats = (
        s.partitions[0].compute().agg(['min', 'max'])
    )

    return (
        partition_0_stats['min'].item(),
        partition_0_stats['max'].item(),
    )

如果需要，您可以利用您对值分布的直觉来扩大此范围：

vmin_p0, vmax_p0 = find_minmax_of_first_partition(df['a'])
range_p0 = (vmax_p0 - vmin_p0)
mean_p0 = (vmin_p0 + vmax_p0) / 2

# guess that the overall data is within 10x the range of partition 1
min_est, max_est = mean_p0 - 5*range_p0, mean_p0 + 5*range_p0

# now, bin all values using this estimated min, max. Note that
# any data falling outside your estimated min/max value will be
# coded as values 0 or K + 1.
bins = np.linspace(min_est, max_est, (K + 1))
binned = s.map_partitions(
    np.digitize,
    bins=bins,
    meta=('binned', 'uint16'),
)

these bins will be equally spaced, but will not necessarily start/end at the min/max and therefore may either not catch all the data or may have empty bins at the edges. You may need to take a look at how your bin specification performs and iterate based on your data.