Многопроцессорная Dask Pymongo - PullRequest
Новая
       8

Многопроцессорная Dask Pymongo

0 голосов
/ 01 апреля 2020

Я пытаюсь использовать dask для параллельного запроса к базе данных MongoDB, но многопроцессорная обработка, похоже, не работает.

У меня есть следующий пример с задержкой 

@dask.delayed
def _retrieve_one(query, settings, new_chunck):
    with MongoClient(settings.MONGODB, settings.MONGOPORT) as client:
        db = client[settings["MONGO_DATABASE"]]
        collection = db[settings["MONGO_COLLECTIONS"]]

        new_query = dict()
        new_query["_id"] = {"$in": new_chunck}

        iterator = collection.find(new_query)

    df_res = pd.concat([pd.DataFrame(x) for x in iterator], axis=0)

    df_res = df_res.reset_index()

    COL = ["ip", "host", "day", "http_method", "start_date"]
    to_concatenate = df_res.loc[:, ["_id"] + COL].drop_duplicates()

    df_res = df_res.drop(COL, axis=1)
    df_res = df_res.pivot(index="_id", columns="index", values="values")
    df_res = df_res.merge(to_concatenate, on="_id")
    return(df_res)

, который я запускаю с этим кодом 

 with MongoClient(self.settings.MONGODB, self.settings.MONGOPORT) as client:
        db = client[self.settings["MONGO_DATABASE"]]
        collection = db[self.settings["MONGO_COLLECTIONS"]]

        # retrieve all id matching query - For multiprocessing
        all_id =\
            list(collection.find(query)\
                 .distinct('_id')
                 )

    logging.info("{} documents to retrieve".format(len(all_id)))
    id_chunck = [all_id[i:i+chunck_size] for i in range(0, len(all_id), chunck_size)]

    dd_df = dd.from_delayed([_retrieve_one(query, self.settings, chunck) for chunck in id_chunck], 
                            meta=pd.DataFrame({"_id": ["a"], 
                                                "bytes":["a"],
                                                "bytes_in":["a"],
                                                "bytes_out":["a"],
                                                "n_queries":["a"],
                                                "ip": ["a"], 
                                                "host": ["a"], 
                                                "day": [datetime.strptime("2020-01-01", '%Y-%m-%d')], 
                                                "http_method": ["a"],
                                                "start_date": [datetime.strptime("2020-01-01", "%Y-%m-%d")]
                                                }))

Это работает (без ошибок), но когда я изменяю chunk_size на несколько разделов, это не происходит быстрее, и кажется, что выполнение происходит только на одном ядре.

1 Ответ

1 голос
/ 04 апреля 2020

Я рекомендую профилировать ваши вычисления, чтобы увидеть, что медленно.

Мое первое предположение - вы связаны сетью или самой базой данных Mon go, но я не знаю.

Для получения дополнительной информации о понимании производительности при использовании Dask см. Следующие документы: https://docs.dask.org/en/latest/understanding-performance.html

...