如何使用 MongoEngine 进行批量插入？答案

【问题标题】：How to do a Bulk Insert with MongoEngine?如何使用 MongoEngine 进行批量插入？
【发布时间】：2018-06-10 05:48:42
【问题描述】：

我正在努力使用 MongoEngine 进行批量插入。我可以轻松地使用循环执行 .save() 以写入数据库，但是我无法从 MongoEngine 文档中找到执行批量插入的过程。唯一提到的是这里：multi document insert using mongoengine into mongodb。我尝试这样做，但我只是插入所有文档而不在架构中验证它们。

这是我的代码：

import mongoengine as me

# connect to MongoDB
me.connect(host='CONNECTION_STRING')

# venue Schema
class newVenues(me.Document):
  name = me.StringField(required=True)
  latitude = me.DecimalField(required=True)
  longitude = me.DecimalField(required=True)


# some test data
venues = [{'name': 'Vega',
           'latitude': 55.672867,
           'longitude': 12.473692},
          {'name': 'Tivoli',
           'latitude': 55.681256,
           'longitude': 12.553412}]

# the list of venues to bulk insert to MongoDB
venues_to_insert_list = []
for venue in venues:
    venues_to_insert_list.append(
    newVenues(
    name=venue.get('name'),
    latitude=venue.get('latitude'),
    longitude=venue.get('longitude')
  )
)

# bulk insert to MongoDB
newVenues.objects.insert(venues_to_insert_list)

【问题讨论】：

你看过.insertMany()
是的，但我不确定如何使用 .insertMany() 或 PyMongo，我猜它是 .insert_many()。我尝试这样做，但出现此错误：TypeError: document must be an instance of dict, bson.son.SON, bson.raw_bson.RawBSONDocument, or a type that inherits from collections.MutableMapping.

标签： python mongodb mongoengine

【解决方案1】：

如果您想插入所有文档而不在 Schema 中验证它们。您需要在 mongoengine 中创建动态文档

以下代码用于读取 csv 文件并在不验证模式的情况下将其插入 mongodb 集合中[模式也是以动态方式生成的]。

import mongoengine as me
class DynamicDoc(me.DynamicDocument):
    any_field = me.StringField()

import pandas as pd

all_csv_records = data_frame.to_dict('records')
data_frame = pd.read_csv(file_path)
for data in all_csv_records:
    report_data = DynamicDoc()
    DynamicDoc.any_field = str('temp_data')
    for col, row in data.iteritems():
        report_data[col] = row
    report_data.save()

注意 - 只有 mongo 模型中的 'any_field' 是必要的以字符串形式输入 manullay。因为字段在 mongo 模型中手动定义。

将 id mongodb 保存为：

【讨论】：

【解决方案2】：

import mongoengine as me

# connect to MongoDB
me.connect(host='CONNECTION_STRING')

# venue Schema
class newVenues(me.Document):
    name = me.StringField(required=True)
    latitude = me.DecimalField(required=True)
    longitude = me.DecimalField(required=True)


# venue Schema
class newVenues(Document):
    name = StringField(required=True)
    latitude = DecimalField(required=True)
    longitude = DecimalField(required=True)


# some test data
venues = [
    {"name": "Vega", "latitude": 55.672867, "longitude": 12.473692},
    {"name": "Tivoli", "latitude": 55.681256, "longitude": 12.553412},
]

# the list of venues to bulk insert to MongoDB


venues_to_insert_list = [newVenues(**data) for data in venues]

# bulk insert to MongoDB
newVenues.objects.insert(venues_to_insert_list)

【讨论】：

这个答案只是重新格式化了 OP 的代码，但用列表理解替换了循环。它没有回答如何进行批量插入 with 架构验证的问题。