dask вычисление не выполняется параллельно

Question

У меня есть каталог json-файлов, которые я пытаюсь преобразовать в dask DataFrame и сохранить в castra. Между ними 200 файлов, содержащих O(10**7) записей json. Код очень прост, в основном следуя учебным примерам.

import dask.dataframe as dd
import dask.bag as db
import json
txt = db.from_filenames('part-*.json')
js = txt.map(json.loads)
df = js.to_dataframe()
cs=df.to_castra("data.castra")

Я использую его на 32-ядерном компьютере, но код использует только одно ядро ​​на 100%. Насколько я понимаю из документов, этот код выполняется параллельно. Почему нет? Я что-то не так понял?

Answer 1

Ваша последняя коллекция - это dask dataframe, который по умолчанию использует потоки, вам придется явно указать dask на использование процессов.

Вы можете сделать это глобально

import dask
import dask.multiprocessing
dask.set_options(get=dask.multiprocessing.get)

Или сделать это только на to_castra вызов

df.to_castra("data.castra", get=dask.multiprocessing.get)

Кроме того, как предупреждение, Кастра была в основном экспериментом. Это прилично быстро, но также и не совсем зрелый, как HDF5.