Как запустить параллельное задание в Python следующим образом?

Question

Я хотел бы иметь возможность выполнять задачу, подобную той, которая показана на рисунке. Функция, которая:

1. Считывает набор данных
2. Применяет некоторые преобразования
3. Параллельно выполняет экспорт набора данных в csv, и пока csv сохраняется, возвращает набор данных в виде pd.DataFrame

Answer 1

Если я правильно понимаю ваш вопрос, этот код для вас:

import pandas as pd
from multiprocessing import Lock, Process
from time import time

def writefile(df,lock, filename):
    lock.acquire()
    df.to_csv(filename, index=False, mode='a', header=False)
    lock.release()


if __name__ == '__main__':
    N = 10000000

    df = pd.DataFrame({'a':range(1, N),'b':range(1, N),'c':range(1, N)})
    filename= "tmp.csv"

    start = time()
    df.to_csv(filename, index=False, mode='a', header=False)
    print("Standard execution time:", time() - start, 'seconds')

    start = time()
    lock = Lock()
    p = Process(target=writefile, args=(df,lock, filename))
    p.start()
    p.join()
    print("Multiprocessing execution time:", time() - new, 'seconds')

Использование многопроцессорной обработки потребует больше времени, чем использование по умолчанию.Используя Синхронизацию между процессами, используйте Процессы и Блокировка для параллельного процесса записи.

Answer 2

Вы можете сделать это:

import threading

def thread_function(df):
    df.to_csv()

def blue_function(df):
    thread = threading.Thread(target=thread_function, args=(df,))
    thread.start()

Из документа не нужно вызывать thread.join(), чтобы дождаться его завершения, так как:

основной потокне поток демона, и поэтому все потоки, созданные в основном потоке, по умолчанию имеют значение daemon = False.Вся программа Python завершается, когда не осталось живых потоков, не являющихся демонами.

РЕДАКТИРОВАТЬ:

При этом вы создаете новый Thread для вашего процесса.Это позволяет ОС самостоятельно планировать время выполнения процедур.Преимущество этого состоит в том, что одному потоку не нужно ждать завершения другого.Таким образом, делая ваш код асинхронным , а не параллельным.

В других языках программирования он также позволяет ОС планировать потоки для запуска на разных ядрах ЦП, заставляя их работать параллельно.Это, однако, невозможно в Python из-за GIL , который блокирует запуск Python более чем одного экземпляра Interpreter одновременно.

Вы можете начать новый процесс вместо потокаОднако это только увеличит накладные расходы (время и память) и не даст никаких преимуществ.Chrome порождает процессы вместо потоков для вкладок браузера, но это происходит из соображений безопасности.Поскольку процессы не разделяют динамическую память.

Если вам действительно требуется параллельное выполнение задач, единственным вариантом является кодирование потоковой части в C и , вызываемых из Python.