Wie durch Spalten mit variabler Bins in Pandas/Dask

Question

Ich bin die Lage, großen Datenrahmen diskretisieren einem Pandas Datenrahmen durch Spalten mit diesem Code diskretisieren:

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 

ich jede Spalte als Parameter diskretisieren will: loc (der Median der Säule) und der Maßstab (median absolute deviation der Säule).

Bei kleinen Datenrahmen ist die benötigte Zeit akzeptabel (da es sich um eine Ein-Thread-Lösung handelt).

Bei größeren Datenrahmen möchte ich jedoch mehr Threads (oder Prozesse) ausnutzen, um die Berechnung zu beschleunigen.

Ich bin kein Experte von Dask, die die Lösung für dieses Problem bieten sollte.

in meinem Fall sollte jedoch die Diskretisierung mit dem Code möglich sein:

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 

aber das Problem hier ist, dass loc und scale auf Spaltenwerte abhängig sind, so sollten sie für jede Spalte berechnet werden, entweder vor oder während der Bewerbung.

Wie kann es gemacht werden?

Answer 1

Ich habe nie dask verwendet, aber ich denke, Sie können eine neue Funktion definieren, die in apply verwendet werden soll.

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