我們有一個包含大約 2000 萬條記錄的資料庫，其中包含索引欄位order_id。
每天，每個小時后，我們都會收到增量更新，從 ~2K 開始增長到 ~50K 記錄，其中一些可能是新的，而另一些則是對以前記錄的更新。
為了處理這些資料，我使用 Airflow 創建了一個管道。在將資料推送到 MongoDB 之前，資料可以作為 Pandas 資料框使用，因為我正在使用它來處理和清理資料。
現在，我正在使用此代碼來更新資料。但我不確定這是否是正確或有效的解決方案。
from pymongo import ReplaceOne
col = MongoDB.get_collection("collection", mongo_db="database")
# My processed data
df
df.reset_index()
bulk_data = [
ReplaceOne({"order_id": row["order_id"]}, dict(row), upsert=True)
for index, row in df.iterrows()
]
col.bulk_write(bulk_data)
那么其他選擇可能是什么？這種做任務的方式是合乎邏輯的還是我做錯了？
uj5u.com熱心網友回復：
你有一種最有效的技術；您的查詢欄位已編入索引，并且您正在使用批量操作。如果它運行緩慢，即使在 50k 記錄上，我也會感到驚訝。
如果您想擠出最后一滴性能，這種方法可能會更快。它洗掉所有傳入的記錄并重新插入它們；通常這比使用ReplaceOne()with更高效upsert=True。to_dict()與使用相比，pandas 中的方法也洗掉了一個步驟iterrows()。最后，您可以將批量排序選項設定為 False，這又是一個性能提升，因為您可能不關心插入的順序。
col.delete_many({"order_id": {'$in': list(df["order_id"])}})
bulk_data = [InsertOne(row) for row in df.to_dict(orient='records')]
col.bulk_write(bulk_data, ordered=False)
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/470075.html
標籤：python-3.x 熊猫 mongodb 空气流动 pymongo
上一篇：無法安裝mongonode.js