pandas gelten multiprocessing

Ich versuche, multiprocessing mit pandas dataframe, ist split der dataframe zu 8 Teile. wendet eine Funktion auf jedes Teil mit apply (mit jedem Teil verarbeitet in verschiedenen Verfahren).

BEARBEITEN:
Hier ist die Lösung, die ich schließlich gefunden:

import multiprocessing as mp
import pandas.util.testing as pdt

def process_apply(x):
    # do some stuff to data here

def process(df):
    res = df.apply(process_apply, axis=1)
    return res

if __name__ == '__main__':
    p = mp.Pool(processes=8)
    split_dfs = np.array_split(big_df,8)
    pool_results = p.map(aoi_proc, split_dfs)
    p.close()
    p.join()

    # merging parts processed by different processes
    parts = pd.concat(pool_results, axis=0)

    # merging newly calculated parts to big_df
    big_df = pd.concat([big_df, parts], axis=1)

    # checking if the dfs were merged correctly
    pdt.assert_series_equal(parts['id'], big_df['id'])
  • es ist ein Raum in der res = df.apply(process apply, axis=1), ist das richtig?
  • was Sie genau zu erreichen versucht, indem dieser code?
  • derzeit gelten nur gesättigte Fettsäuren einen Kern der CPU. Ich will Multiprozess-und nutzen alle Kerne zu verringern Bearbeitungszeit
  • danke für den Hinweis, aber das ist nur ein Tippfehler im post
  • Es wäre schöner, wenn man die Frage allein und setzen Sie dann die Antworten in den Antworten. So können wir sehen, mehr von dem Prozess, ohne Blick auf das changelog.
  • sollte "aoi_proc" sein "Prozess"? Vielleicht ist die Umbenennung der "Prozess" - Funktion, um einfach "f" wäre besser lesbar in der multiprocessing-Kontext

InformationsquelleAutor yemu | 2014-11-06
  1. 9

    Einer allgemeineren version, basierend auf der Autor-Lösung, die es erlaubt, führen Sie es auf jede Funktion und dataframe:

    from multiprocessing import  Pool
from functools import partial
import numpy as np

def parallelize(data, func, num_of_processes=8):
    data_split = np.array_split(data, num_of_processes)
    pool = Pool(num_of_processes)
    data = pd.concat(pool.map(func, data_split))
    pool.close()
    pool.join()
    return data

def run_on_subset(func, data_subset):
    return data_subset.apply(func, axis=1)

def parallelize_on_rows(data, func, num_of_processes=8):
    return parallelize(data, partial(run_on_subset, func), num_of_processes)

    Also die folgende Zeile:

    df.apply(some_func, axis=1)

    Werden:

    parallelize_on_rows(df, some_func)
    InformationsquelleAutor Tom Raz
  2. 4

    Da habe ich nicht viel von Ihrer Daten-Skript, dies ist eine Vermutung, aber ich würde vorschlagen, mit p.map statt apply_async mit dem Rückruf.

    p = mp.Pool(8)
pool_results = p.map(process, np.array_split(big_df,8))
p.close()
p.join()
results = []
for result in pool_results:
    results.extend(result)
    • hat diese Arbeit für Sie?
    • Ich musste die call-in, wenn name == 'main'. und mit anderen kleinen änderungen habe ich es geschafft, damit es funktioniert, aber ich bin mir nicht sicher, ob das Ergebnis dataframes im pool Ergebnisse werden in der gleichen Reihenfolge zurückgegeben, wie Sie waren gespalten. Ich habe um es zu überprüfen.
    • siehe hier für eine Lösung mit dask stackoverflow.com/questions/37979167/...
    InformationsquelleAutor Rafael Barros
  3. 1

    Können Sie https://github.com/nalepae/pandarallel, wie im folgenden Beispiel:

    from pandarallel import pandarallel
from math import sin

pandarallel.initialize()

def func(x):
    return sin(x**2)

df.parallel_apply(func, axis=1)
    InformationsquelleAutor Sébastien Vincent
  4. 0

    Ich auch das gleiche problem, wenn ich multiprocessing.map() zu bewerben-Funktion zu anderen chunk von einem großen dataframe.

    Ich möchte nur hinzufügen mehrere Punkte, nur für den Fall, andere Menschen laufen in das gleiche problem wie ich.

    1. erinnern hinzufügen if __name__ == '__main__':
    2. führen Sie die Datei in einem .py - Datei, wenn Sie ipython/jupyter notebook, dann können Sie nicht laufen multiprocessing (das gilt für meinen Fall, obwohl ich keine Ahnung habe)
    InformationsquelleAutor user6651227
  5. 0

    Dies funktionierte gut für mich:

    rows_iter = (row for _, row in df.iterrows())

with multiprocessing.Pool() as pool:
    df['new_column'] = pool.map(process_apply, rows_iter)
    InformationsquelleAutor EliadL
Schreibe einen Kommentar