Графы задач и методы планирования

Параллельное программирование с Dask на Python

James Fulton

Climate Informatics Researcher

Визуализация графа задач

# Create 2 delayed objects
delayed_num1 = delayed(my_square_function)(3)
delayed_num2 = delayed(my_square_function)(4)

# Add them
result = delayed_num1 + delayed_num2


# Plot the task graph
result.visualize()

Диаграмма, показывающая шаги для вычисления результата. my-square-function выполняется дважды, и два результата передаются в функцию сложения, которая возвращает одно значение.

Пересекающийся граф задач

delayed_intermediate = delayed(my_square_function)(3)

# These two results both use delayed_intermediate_result
delayed_result1 = delayed_intermediate - 5
delayed_result2 = delayed_intermediate + 4

Пересекающийся граф задач

delayed_result1.visualize()

Диаграмма, показывающая граф задач для результата 1.

delayed_result2.visualize()

Диаграмма, показывающая граф задач для результата 2.

Пересекающийся граф задач

# Plot the task graph
dask.visualize(delayed_result1, delayed_result2)

Граф задач, показывающий, что результат 1 и результат 2 используют общий промежуточный результат.

Многопоточность и параллельная обработка

Перемещение данных

Параллельная обработка

Процессы имеют собственное пространство оперативной памяти

Многопоточность

Потоки используют общее пространство оперативной памяти

Многопоточность и параллельная обработка

# Run a sum on two big arrays
sum1 = delayed(np.sum)(big_array1)
sum2 = delayed(np.sum)(big_array2)

# Compute using processes
dask.compute(sum1, sum2)

При параллельной обработке работает медленно

Диаграмма показывает, что два массива из одного процесса Python необходимо передать двум другим процессам Python.

Многопоточность и параллельная обработка

# Run a sum on two big arrays
sum1 = delayed(np.sum)(big_array1)
sum2 = delayed(np.sum)(big_array2)

# Compute using threads
dask.compute(sum1, sum2)

При многопоточности работает быстро

Диаграмма показывает, что два массива не нужно копировать вовсе.

GIL

Глобальная блокировка интерпретатора — только один поток может читать скрипт Python в каждый момент времени

def sum_to_n(n):
    """Sums numbers from 0 to n"""
    total = 0
    for i in range(n+1):
        total += i
    return total

Многопоточность здесь не поможет
Параллельная обработка поможет

sum1 = delayed(sum_to_n)(1000)
sum2 = delayed(sum_to_n)(1000)

Пример замеров времени — GIL

Три диаграммы Ганта, показывающие время выполнения простой функции Python 16 раз. Из трёх методов планирования задач параллельная обработка оказалась самой быстрой.

Функции, снимающие блокировку GIL

Например, функция pd.read_csv() снимает блокировку GIL

df1 = delayed(pd.read_csv)('file1.csv')
df2 = delayed(pd.read_csv)('file2.csv')

Пример замеров времени — загрузка данных

Итоги

Потоки

Запускаются очень быстро
Используют общую память с основным сеансом
Не требуют передачи данных в памяти
Ограничены блокировкой GIL: только один поток может читать код одновременно

Процессы

Требуют времени и памяти на запуск
Имеют раздельные пулы памяти
Очень медленно обмениваются данными между собой и с основным сеансом Python
Каждый имеет собственный GIL и не конкурирует с другими за чтение кода

Давайте потренируемся!

Параллельное программирование с Dask на Python