使用插入和更新每日刷新 MongoDB 集合

Question

我有一个 MongoDB 集合，其中需要每晚刷新某些字段的数据。 Target 集合有 3 个额外的自定义字段，供最终用户输入相应文档时使用。

因此，当每日刷新在一夜之间发生时，数据源可以发送新文档或现有文档的更新数据。文档最多可达 10,000 个。

我正在使用 Pymongo 和 MongoDB 来实现这一点。我的问题是，如何在不影响最终用户数据的情况下识别需要更新哪些记录以及需要插入哪些记录以及这 3 个额外的自定义字段。

例如：

数据来源：

Manufacture Name       Model        Year    Units
BMW                    5Series      2019      10
BMW                    5Series      2020       5
AUDI                   A4           2020      20 
AUDI                   A7           2019       3
TOYOTA                 COROLLA      2020       5
TOYOTA                 CAMRY        2020       6
HONDA                  ACCORD       2020       10
HONDA                  PILOT        2019       15  
HONDA                  CRV          2019       20 

加载后，App 表有 1 个自定义列（位置）供用户输入

Manufacture Name       Model        Year     Location   Units
BMW                    5Series      2019     London       10 
BMW                    5Series      2020     New York     5  
AUDI                   A4           2020     Melbourne    20
AUDI                   A7           2019     London       3
TOYOTA                 COROLLA      2020     New York     5
TOYOTA                 CAMRY        2020     London       6  
HONDA                  ACCORD       2020     Sydney       10 
HONDA                  PILOT        2019     Tokyo        15
HONDA                  CRV          2019

第二天，我们得到如下新数据

Manufacture Name       Model        Year    Units
BMW                    5Series      2019      10
BMW                    5Series      2020       **35**
**BMW                    7Series      2020       12**
AUDI                   A4           2020      20 
AUDI                   A7           2019       3
**AUDI                   A6           2019       1**
TOYOTA                 COROLLA      2020       5
TOYOTA                 CAMRY        2020       6
HONDA                  ACCORD       2020       10
HONDA                  PILOT        2019       15  
*HONDA                  CRV          2019       20*       *-- deleted -- in second refresh*

数据可以是 10,000 条记录。如何使用 Pymongo 或 MongodB 实现这一点？我在 PyMongo 中编写代码，直到检索源数据并将光标存储在字典中。不确定在此之后如何使用 MongoDB Upsert 或批量写入和保留/更新现有记录的位置列数据并为新记录分配 NULL 值。

谢谢

Answer 1

最终实现如下：

import pymongo
from pymongo import UpdateOne


# Define MongoDB connection
client = pymongo.MongoClient(server, username='User', password='password', authSource='DBName', authMechanism='SCRAM-SHA-256')

#Source table    
collection2 = db['cars2']

count123 = collection2.count_documents({})

#print("New Cars2 Data - Count before Insert:",count123)

source_cursor = collection2.find()

print("Cars2 - Count in Cursor:",source_cursor.count())

#Target Table    
collection = db['cars']

tgt_count = collection.count_documents({})

print("Cars Collection - Count before Insert:",tgt_count)

sourcedata = []

#since this is a MongoDB Cursor object, we need push to List using list()
sourcedata = list(source_cursor)

source_cursor.close()



# ADD new columns to the Data before inserting in MongoDB
for item in sourcedata:    

    item['Location'] = None        
    item['last_refresh'] =  datetime.now()


#sourcedata = source_cursor

ops = []

if tgt_count == 0:

    print("Loading for First time:")
    for rec in sourcedata:        
        #Load Data for first time with new fields
        ops.append( UpdateOne({'name': rec['name'],'model':rec['model']}, {'$set': {'name':rec['name'],'model':rec['model'],'year':rec['year'],'units':rec['units'], 'Location': rec['Location'],'last_refresh':rec['last_refresh']}}, upsert=True))

    #print(ops)

    result = collection.bulk_write(ops)


    print("Inserted Count:", result.inserted_count)
    print("Matched Count:", result.matched_count)
    print("Modified Count:", result.modified_count)
    print("Upserted Count:", result.upserted_count)

elif tgt_count > 0:

    print("Updating the Load:")
    for rec in sourcedata:        
        #No Location field is included to avoid replacing the values of this field by NULL
        ops.append( UpdateOne({'name': rec['name'],'model':rec['model']}, {'$set': {'name':rec['name'],'model':rec['model'],'year':rec['year'],'units':rec['units'], 'last_refresh':rec['last_refresh']}}, upsert=True))

    result = collection.bulk_write(ops)

    print("Inserted Count:", result.inserted_count)
    print("Matched Count:", result.matched_count)
    print("Modified Count:", result.modified_count)
    print("Upserted Count:", result.upserted_count)

 #because I didn’t include “Location” field above, the new UPSERT records DO NOT have “Location” field anymore. So I have to update the collection one more time to include “Location” field        

    nullfld_result = collection.update_many({'Location':None},{ '$set':{'Location':None}})


count2 = collection.count_documents({})

print("Count after Insert:",count2)