Многопроцессорность Pandas SQL

Я получаю сообщение об ошибке «Таблица не найдена», когда использую многопроцессорную обработку для фрагментированного фрейма данных Pandas в каждом процессоре, запущенном библиотекой mp.
Я использую библиотеку pandasql для SQL следующим образом:

import pandasql import sqldf
pysqldf = lambda q: sqldf(q, globals())

df = pd.DataFrame({'a': [1,2,4,3,6,1,2], 'b': ['a','a','b','b','c','c','c']})

Это работает для одного потока:

sorted_df = pysqldf("select * from df order by b, a")

Это не работает, когда я применяю многопроцессорность для параллельной обработки каждого фрагмента df:

def parallelize_dataframe(df, func):
    unique_bs = df.b.unique().tolist()
    df_split = [df[df.a == l] for l in unique_bs]
    df = pd.concat(pool.map(func, df_split))
    pool = Pool(num_cores)
    pool.close()
    pool.join()
    return df

def sort_chunks(data):
    sorted_data = pysqldf("select * from data order by b, a")
    return sorted_data

sorted_df = parallelize_dataframe(df, sort_chunks)

Ошибка, которую я получаю, следующая:

PandaSQLException: (sqlite3.OperationalError) нет такой таблицы: данные [SQL: 'выбрать * из данных'] (предыстория этой ошибки по адресу: http://sqlalche.me/e/e3q8)

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


python pandas multiprocessing pandasql
person Dan Lee    schedule 03.07.2018    source источник

Ответы (1)


arrow_upward
0
arrow_downward

Я понял. Проблема была в пространстве имен. По сути, pysqldf принимает два аргумента — запрос и пространство имен, которое может быть либо глобальным, либо локальным. Если функция sqldf заключена внутри функции и ей необходимо использовать локальный фрейм данных для этой функции, аргумент пространства имен должен быть локальным.

Изменение функции данных сортировки на приведенную ниже устранило проблему:

def sort_chunks(data):
    sorted_data = pysqldf("select * from data order by b, a", locals())
    return sorted_data
person Dan Lee    schedule 04.07.2018