python dask DataFrame, Unterstützung für (trivial parallelisierbare) Zeile anwenden?

Vor kurzem fand ich dask Modul zielt darauf ab, dass ein einfach zu bedienendes python-parallel-processing-Modul. Großer Pluspunkt für mich ist, dass es funktioniert mit pandas.

Nach der Lektüre ein wenig auf seiner manual-Seite, ich kann nicht einen Weg finden, dies zu tun trivial parallelisierbare Aufgabe: 

ts.apply(func) # for pandas series
df.apply(func, axis = 1) # for pandas DF row apply

Im moment, dies zu erreichen in dask, AFAIK,

ddf.assign(A=lambda df: df.apply(func, axis=1)).compute() # dask DataFrame

die hässliche syntax und ist tatsächlich langsamer als geradezu

df.apply(func, axis = 1) # for pandas DF row apply

Jede Anregung?

Edit: Danke @MRocklin für die map-Funktion. Es scheint, langsamer zu sein als nur pandas gelten. Ist das mit den pandas GIL-releasing-Problem oder mache ich es falsch?

import dask.dataframe as dd
s = pd.Series([10000]*120)
ds = dd.from_pandas(s, npartitions = 3)

def slow_func(k):
    A = np.random.normal(size = k) # k = 10000
    s = 0
    for a in A:
        if a > 0:
            s += 1
        else:
            s -= 1
    return s

s.apply(slow_func) # 0.43 sec
ds.map(slow_func).compute() # 2.04 sec
  • Ich bin nicht vertraut mit dask Modul. Für mulit-processing, python-Modul multiprocessing funktioniert gut für mich, wenn ich zum verarbeiten einer großen dataframe Zeile-für-Zeile. Die Idee ist auch sehr einfach: verwenden Sie np.array_split um split große dataframe in 8 Stücke schneiden und verarbeiten Sie gleichzeitig mit multiprocessing; Sobald es fertig ist, verwenden Sie pd.concat zu concat Sie zurück zu der ursprünglichen Länge. Für einen entsprechenden post mit vollständige Codebeispiel finden Sie unter stackoverflow.com/questions/30904354/...
  • Danke, sehr nett. Das problem der multiprocessing-Modul ist, dass Sie brauchen, um eine benannte Funktion (nicht lambda) und legte es der name=="main" block. Das macht die Forschung-codes schlecht strukturiert ist.
  • Wenn Sie nur wollen, um eine bessere Nutzung multiprocessing kann man sich unter Multiprozess von @mike-mckerns . Sie könnten auch versuchen, aus dask core statt dask.dataframe und erstellen Sie Wörterbücher oder verwenden Sie so etwas wie github.com/ContinuumIO/dask/pull/408
InformationsquelleAutor jf328 | 2015-07-11
  1. 55

    map_partitions

    Können Sie Ihre Funktion, um alle Partitionen Ihres dataframe mit den map_partitions Funktion.

    df.map_partitions(func, columns=...)

    Beachten Sie, dass func wird nur ein Teil des dataset an eine Zeit, nicht den gesamten Datenbestand wie mit pandas apply (die vermutlich Sie würde nicht wollen, wenn Sie wollen, um die Parallelität.)

    map /apply

    Können Sie anzeigen eine Funktion zeilenweise über eine Reihe mit map

    df.mycolumn.map(func)

    Können Sie anzeigen eine Funktion zeilenweise über ein dataframe mit apply

    df.apply(func, axis=1)

    Threads vs. Prozesse

    Ab version 0.6.0 dask.dataframes parallelizes mit threads. Benutzerdefinierte Python-Funktionen werden nicht erhalten viel nutzen von thread-basierte Parallelisierung. Sie könnten versuchen, Prozesse statt

    df = dd.read_csv(...)

df.map_partitions(func, columns=...).compute(scheduler='processes')

    Aber vermeiden apply

    Allerdings sollte man wirklich vermeiden apply mit benutzerdefinierten Python-Funktionen, sowohl in der Pandas und in Dask. Dies ist oft eine Quelle der schlechten Leistung. Könnte es sein, dass, wenn Sie einen Weg finden, um Ihren Betrieb in eine vektorisierte Art und Weise, dann könnte es sein, dass Ihr Pandas code 100x schneller und Sie müssen nicht dask.dataframe an alle.

    Betrachten numba

    Für Ihr spezielles problem könnten Sie in Erwägung ziehen numba. Dies verbessert deutlich die Leistung.

    In [1]: import numpy as np
In [2]: import pandas as pd
In [3]: s = pd.Series([10000]*120)

In [4]: %paste
def slow_func(k):
    A = np.random.normal(size = k) # k = 10000
    s = 0
    for a in A:
        if a > 0:
            s += 1
        else:
            s -= 1
    return s
## -- End pasted text --

In [5]: %time _ = s.apply(slow_func)
CPU times: user 345 ms, sys: 3.28 ms, total: 348 ms
Wall time: 347 ms

In [6]: import numba
In [7]: fast_func = numba.jit(slow_func)

In [8]: %time _ = s.apply(fast_func)  # First time incurs compilation overhead
CPU times: user 179 ms, sys: 0 ns, total: 179 ms
Wall time: 175 ms

In [9]: %time _ = s.apply(fast_func)  # Subsequent times are all gain
CPU times: user 68.8 ms, sys: 27 µs, total: 68.8 ms
Wall time: 68.7 ms

    Disclaimer, ich arbeite für die Firma, die macht beide numba und dask und beschäftigt viele der pandas Entwickler.

    • Danke! Ich habe versucht, die map-Methode, und es scheint, langsamer zu sein als die pandas gelten. Könnten Sie Kommentar Bearbeiten auf der original post bitte?
    • Ich habe bearbeitet die obige Antwort.
    • Ich verwende bereits numba eine Menge! Vielen Dank für die Arbeit. Was ich normalerweise tun, ist, dass jede Zeile des DataFrame gibt eine Konfiguration einer simulation (Parameter für eine komplexe/langsame func). Ich mache multiprocessing schon, einfach ein Auge auf die besseren Möglichkeiten, dies zu tun
    • Etwas off-topic bezüglich pandas; ich versuche, anzeigen über anwenden, weil ich habe gehört, es ist schneller, aber ich bin mir nicht sicher, warum es ist schneller. Keine Klärung oder links zu Klärung wäre sehr dankbar.
    • keine Ahnung. Schlagen Sie ein kleines experiment und die Entsendung einer "warum ist das der Fall" Stil stackoverflow-Frage.
    InformationsquelleAutor MRocklin
  2. 2

    Als der v dask.dataframe.gelten Delegierten Verantwortung zu map_partitions:

    @insert_meta_param_description(pad=12)
def apply(self, func, convert_dtype=True, meta=no_default, args=(), **kwds):
    """ Parallel version of pandas.Series.apply
    ...
    """
    if meta is no_default:
        msg = ("`meta` is not specified, inferred from partial data. "
               "Please provide `meta` if the result is unexpected.\n"
               "  Before: .apply(func)\n"
               "  After:  .apply(func, meta={'x': 'f8', 'y': 'f8'}) for dataframe result\n"
               "  or:     .apply(func, meta=('x', 'f8'))            for series result")
        warnings.warn(msg)

        meta = _emulate(M.apply, self._meta_nonempty, func,
                        convert_dtype=convert_dtype,
                        args=args, **kwds)

    return map_partitions(M.apply, self, func,
                          convert_dtype, args, meta=meta, **kwds)
    InformationsquelleAutor Shubham Chaudhary
Schreibe einen Kommentar