Một cách tốt hơn để tải dữ liệu MongoDB vào một DataFrame bằng cách sử dụng Pandas và PyMongo?

Question

Tôi có một cơ sở dữ liệu MongoDB 0.7 GB chứa các mẩu tin mà tôi đang cố tải vào một khung dữ liệu. Tuy nhiên, tôi gặp lỗi.

MemoryError:  

Mã của tôi trông như thế này:

cursor = tweets.find() #Where tweets is my collection 
tweet_fields = ['id'] 
result = DataFrame(list(cursor), columns = tweet_fields) 

Tôi đã thử các phương pháp trong các câu trả lời sau đây, mà tại một số điểm tạo ra một danh sách tất cả các yếu tố của cơ sở dữ liệu trước khi tải nó.

• https://stackoverflow.com/a/17805626/2297475
• https://stackoverflow.com/a/16255680/2297475

Tuy nhiên, trong một câu trả lời mà nói về danh sách(), người nói rằng đó là tốt cho các tập dữ liệu nhỏ, bởi vì tất cả mọi thứ được nạp vào bộ nhớ.

• https://stackoverflow.com/a/13215411/2297475

Trong trường hợp của tôi, tôi nghĩ rằng đó là nguồn gốc của lỗi. Có quá nhiều dữ liệu được tải vào bộ nhớ. Tôi có thể sử dụng phương pháp nào khác?

Answer 1

Tôi đã sửa đổi mã của tôi như sau:

cursor = tweets.find(fields=['id']) 
tweet_fields = ['id'] 
result = DataFrame(list(cursor), columns = tweet_fields) 

Bằng cách thêm lĩnh vực tham số trong hàm find() Tôi hạn chế đầu ra. Điều đó có nghĩa là tôi không tải mọi trường nhưng chỉ các trường được chọn vào DataFrame. Tất cả mọi thứ hoạt động tốt ngay bây giờ.

Answer 2

Cách nhanh nhất và hiệu quả nhất về bộ nhớ, để tạo Khung dữ liệu từ truy vấn mongodb, như trong trường hợp của bạn, sẽ sử dụng monary.

This post có lời giải thích ngắn gọn và súc tích.

Answer 3

một cách thanh lịch để làm việc đó sẽ như sau:

import pandas as pd 
def my_transform_logic(x): 
    if x : 
     do_something 
     return result 

def process(cursor): 
    df = pd.DataFrame(list(cursor)) 
    df['result_col'] = df['col_to_be_processed'].apply(lambda value: my_transform_logic(value)) 

    #making list off dictionaries 
    db.collection_name.insert_many(final_df.to_dict('records')) 

    # or update 
    db.collection_name.update_many(final_df.to_dict('records'),upsert=True) 


#make a list of cursors.. you can read the parallel_scan api of pymongo 

cursors = mongo_collection.parallel_scan(6) 
for cursor in cursors: 
    process(cursor) 

tôi đã cố gắng quá trình nêu trên trong một bộ sưu tập MongoDB với 2,6 triệu bản sử dụng Joblib trên mã trên. Mã của tôi không ném bất kỳ lỗi bộ nhớ nào và quá trình xử lý hoàn tất sau 2 giờ.