我尝试使用 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 更改为具有多个分区时,它的速度并不快,而且似乎只在一个核心上执行。
我建议分析您的计算以查看什么是缓慢的。
我的第一个猜测是您受网络或 Mongo 数据库本身的约束,但我不知道。
有关在使用 Dask 时了解性能的更多信息,请参阅这些文档:https ://docs.dask.org/en/latest/understanding-performance.html
本文收集自互联网,转载请注明来源。
如有侵权,请联系 [email protected] 删除。
我来说两句