2

我可以使用以下代码按列离散化 Pandas 数据帧：

import numpy as np
import pandas as pd

def discretize(X, n_scale=1):

    for c in X.columns:
        loc = X[c].median()

        # median absolute deviation of the column
        scale = mad(X[c])

        bins = [-np.inf, loc - (scale * n_scale),
                loc + (scale * n_scale), np.inf]
        X[c] = pd.cut(X[c], bins, labels=[-1, 0, 1])

    return X

我想使用作为参数对每一列进行离散化：loc（列的中值）和 scale（列的中值绝对偏差）。

对于小型数据帧，所需的时间是可以接受的（因为它是单线程解决方案）。

但是，对于更大的数据帧，我想利用更多的线程（或进程）来加速计算。

我不是Dask的专家的专家，它应该为这个问题提供解决方案。

但是，在我的情况下，代码的离散化应该是可行的：

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

def discretize(X, n_scale=1):

    # I'm using only 2 partitions for this example
    X_dask = dd.from_pandas(X, npartitions=2)

    # FIXME: 
    # how can I define bins to compute loc and scale
    # for each column?
    bins = [-np.inf, loc - (scale * n_scale),
            loc + (scale * n_scale), np.inf]

    X = X_dask.apply(pd.cut, axis=1, args=(bins,), labels=[-1, 0, 1]).compute()

    return X

loc但这里的问题是scale依赖于列值，因此应该在应用之前或期间为每一列计算它们。

怎么做到呢？

4

1 回答 1

1

我从来没有用过dask，但我想你可以定义一个新的函数来使用apply。

import dask.dataframe as dd
import multiprocessing as mp
import numpy as np
import pandas as pd

def discretize(X, n_scale=1):

    X_dask = dd.from_pandas(X.T, npartitions=mp.cpu_count()+1)
    X = X_dask.apply(_discretize_series,
                     axis=1, args=(n_scale,),
                     columns=X.columns).compute().T

    return X

def _discretize_series(x, n_scale=1):

    loc = x.median()
    scale = mad(x)
    bins = [-np.inf, loc - (scale * n_scale),
            loc + (scale * n_scale), np.inf]
    x = pd.cut(x, bins, labels=[-1, 0, 1])

    return x
于 2016-08-09T02:57:04.883 回答