Dask-Dataframe 列中唯一值的数量答案

【问题标题】：Number of Unique values in Dask-Dataframe columnsDask-Dataframe 列中唯一值的数量
【发布时间】：2021-12-23 23:13:36
【问题描述】：

我有一个从 csv 文件中读取的 Dask Dataframe，其中包含大约 100 万条记录和 120 个特征/列，我想计算每列中唯一值的数量。我可以清楚地使用 for 循环分别为每一列执行此操作：

from dask import dataframe as dd
dask_df = dd.read_csv("train.csv")
for column in dask_df.columns:
     print(dask_df[col].nunique().compute())

但是每次迭代的计算都非常昂贵（我花了大约 40 分钟，3 个节点集群有 5 个工作人员，每个工作人员有 2GB 内存和 2 个 vcore），所以有没有一种方法可以让我在每个地方获得唯一值数据框列？我试过 dask_df.describe() api，但它只为字符串类型提供唯一值。任何帮助表示赞赏，在此先感谢！

【问题讨论】：

标签： python dask dask-distributed dask-dataframe

【解决方案1】：

我不知道这是否是最快的解决方案，但您可以使用 .melt() 取消透视您的数据框列，然后在变量列上使用 .groupby() 来计算每个组中的唯一值以获得显着性每列解决方案的性能改进：

dd.read_csv('test.csv').melt().groupby('variable')['value'].nunique().compute()

让我们生成一些随机整数数据并保存为 csv：

import numpy as np
import pandas as pd
from dask import dataframe as dd

nrows = 10000
ncols = 120

rng = np.random.default_rng(seed=1)
random_data = rng.integers(low=0, high=nrows/2, size=(nrows,ncols))
pd.DataFrame(data=random_data).add_prefix('col_').to_csv('test.csv', index=False)

我们使用以下两个函数进行性能评估：

def nunique_per_column():
    dask_df = dd.read_csv('test.csv')
    counts = []
    for col in dask_df.columns:
          counts.append(dask_df[col].nunique().compute())
    return pd.Series(counts, index=dask_df.columns)

def melt_groupby_nunique():
    return dd.read_csv('test.csv').melt().groupby('variable')['value'].nunique().compute()

首先检查两个函数是否计算相同的结果：

pd.testing.assert_series_equal(nunique_per_column().sort_index(),
                               melt_groupby_nunique().sort_index(),
                               check_names=False)

%timeit 函数和示例数据在我的机器上产生以下输出：

%timeit nunique_per_column()
17.5 s ± 216 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

%timeit melt_groupby_nunique()
1.78 s ± 576 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

【讨论】：

当我尝试使用melt()运行时，似乎worker消耗了大量内存但性能略有提升，使用dask_df.compute().nunique()怎么样，因为这样更快比所有这些（大约需要 30 秒）但是 compute() 将需要整个数据帧来适应内存？
如果您的数据帧适合内存，您可以在调用 compute() 后使用 pd.DataFrame.nunique()。但是 nunique() 不适用于 dask 数据帧，您可以使用 groupby。也看看this question。

【解决方案2】：

这是另一种解决方法，每列的唯一值的数量都是一次计算的，从而有更多的优化机会：

import random
import pandas
import dask
import dask.dataframe as dd

df = pandas.DataFrame({
    "x": [random.randint(0,100) for _ in range(100)],
    "y": [random.randint(0,100) for _ in range(100)],
    "z": [random.randint(0,100) for _ in range(100)],
})
ddf = dd.from_pandas(df, npartitions=10)

unique = {
    name: ddf[name].nunique()
    for name in ddf.columns
}

# traverse=True is default, but being explicit that we are traversing the dict for dask objects
dask.compute(unique, traverse=True)

【讨论】：

【解决方案3】：

@Mohamed 在 Dask 版本中 2022.01.0，dask.DataFrame.nunique() 一直是 implemented：

import random
import pandas
import dask.dataframe as dd

df = pandas.DataFrame({
    "x": [random.randint(0,100) for _ in range(100)],
    "y": [random.randint(0,100) for _ in range(100)],
    "z": [random.randint(0,100) for _ in range(100)],
})
ddf = dd.from_pandas(df, npartitions=10)

ddf.nunique().compute()

【讨论】：

这很酷，会检查并投票！