python - Pandas/Dask で可変ビンを使用して列ごとに大きなデータフレームを離散化する方法

Question

次のコードを使用して、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が列の値に依存しているため、適用前または適用中に各列に対して計算する必要があることです。

どうすればそれができますか？