将 lambda 函数应用于 dask 数据帧

Question

我希望将一个lambda函数应用于 dask 数据框以更改列中的标签，如果它小于某个百分比。我使用的方法适用于 Pandas 数据帧，但相同的代码不适用于 dask 数据帧。代码如下。

df = pd.DataFrame({'A':['ant','ant','cherry', 'bee', 'ant'], 'B':['cat','peach', 'cat', 'cat', 'peach'], 'C':['dog','dog','roo', 'emu', 'emu']})

ddf = dd.from_pandas(df, npartitions=2)

df:

输出：

     A     B      C

0   ant    cat   dog

1   ant    peach dog

2   cherry cat   roo

3   bee    cat   emu

4   ant    peach emu

ddf.compute()

输出：

     A     B      C

0   ant    cat   dog

1   ant    peach dog

2   cherry cat   roo

3   bee    cat   emu

4   ant    peach emu

list_ = ['B','C']

df.apply(lambda x: x.mask(x.map(x.value_counts(normalize=True))<.5, 'other') if x.name not in list_ else x)

输出：

     A     B      C

0   ant    cat   dog

1   ant    peach dog

2   other  cat   roo

3   other  cat   emu

4   ant    peach emu

对 dask 数据框执行相同操作：

ddf.apply(lambda x: x.mask(x.map(x.value_counts(normalize=True))<.5, 'other') if x.name not in list_ else x,axis=1).compute()

输出（给出警告而不是所需的输出）：

/home/michael/env/lib/python3.5/site-packages/dask/dataframe/core.py:3107: UserWarning: `meta` is not specified, inferred from partial data. Please provide `meta` if the result is unexpected.

  Before: .apply(func)

  After:  .apply(func, meta={'x': 'f8', 'y': 'f8'}) for dataframe result

  or:     .apply(func, meta=('x', 'f8'))            for series result

  warnings.warn(msg)

      A       B       C

0   other   other   other

1   other   other   other

2   other   other   other

3   other   other   other

4   other   other   other

有人可以帮助我获得 dask 数据帧实例所需的输出。

Answer 1

在 pandas 和 dask 情况下，您没有执行相同的操作：对于后者，您有axis=1，因此您最终会替换给定行中出现少于两次的任何值，即所有这些值。如果您更改为axis=0，您将看到您收到异常。这是因为要计算第一个分区，您还需要将整个数据帧传递给 lambda 函数 - 否则如何获得 value_counts？您的问题的解决方案是分别获取值计数。您可以显式地计算它（结果很小）或将其传递给 lambda。此外请注意，走这条路意味着您可以避免使用apply赞成map和使事情更加明确。在这里我专门选择一列，你可以循环。vc = ddf.A.value_counts().compute()vc /= vc.sum()&nbsp; # because dask's value_count doesn't normalisedef simple_map(df):&nbsp; &nbsp; df['A'] = df['A'].map(lambda x: x if vc[x] > 0.5 else 'other')&nbsp; &nbsp; return dfddf.map_partitions(simple_map, meta=df[:0]).compute()