如何通过MongoEngine批量导入JSON数据至MongoDB？

问题背景 #

尝试将Pandas DataFrame中的大量数据通过MongoEngine导入MongoDB，先将DataFrame转为JSON格式，执行导入时出现TypeError: __main__.MarketData() argument after ** must be a mapping, not str错误。

现有代码及问题 #

1. DataFrame转JSON代码 #

result = df.to_json(orient="index")
parsed = json.loads(result)
json_data = json.dumps(parsed, indent=4) 

2. JSON格式化代码 #

json_object = json.loads(json_data)
json_formatted_str = json.dumps(json_object, indent=2)
print(json_formatted_str)

得到的JSON结构：

{
  "0": {
    "Address": " Bursvej 30 ",
    "Zip/city": "4930 Maribo",
    "Price": " 148.000kr. ",
    "Date": 1673371545635
  },
  "1": {
    "Address": " Garrdesmuttevej 20 ",
    "Zip/city": "9550 Mariager",
    "Price": " 148.000kr. ",
    "Date": 1673371545635
  },
  "2": {
    "Address": " Norrevej 21 ",
    "Zip/city": "6990 Ulfborg",
    "Price": " 150.000kr. ",
    "Date": 1673371545635
  }
}

3. 导入代码及错误 #

MD = [MarketData(**data) for data in json_formatted_str]
MarketData.objects.insert(MD, load_bulk=False)

错误信息：TypeError: __main__.MarketData() argument after ** must be a mapping, not str

4. DataFrame示例 #

Address               Zip/city        Price               Date
0   Bursøvej 30, Bursø    4930 Maribo    148.000 kr.    2023-01-10 17:25:45.635483
1   Gærdesmuttevej 20     9550 Mariager   148.000 kr.    2023-01-10 17:25:45.635483
2   Nørrevej 21           6990 Ulfborg    150.000 kr.    2023-01-10 17:25:45.635483
3   Egernvænget 54        4733 Tappernøje 195.000 kr.    2023-01-10 17:25:45.635483
4   Egernvænget 56        4733 Tappernøje 195.000 kr.    2023-01-10 17:25:45.635483

5. MarketData模型代码 #

class MarketData(Document):
    #answers = DictField()
    Address = DynamicField(required=False)
    city = DynamicField(required=False)
    Price = DynamicField(required=False)
    date = DynamicField(required=False)
    
    def json(self):
        market_dict = {
            "username": self.username,
            "city": self.city,
            "Price": self.price
        }
        return json.dumps(market_dict)

问题原因 #

错误的核心是json_formatted_str是字符串类型，遍历字符串会得到单个字符，导致MarketData(**data)中的data是单个字符而非字典映射。另外，现有代码存在冗余的JSON序列化/反序列化步骤，且DataFrame字段与模型字段不完全匹配（如模型中的city对应DataFrame的Zip/city，date对应Date）。

解决方法 #

方法一：直接从DataFrame生成MongoEngine对象列表（推荐） #

跳过冗余的JSON转换步骤，直接从DataFrame的行数据生成对象：

# 处理DataFrame字段与模型字段的映射
records = df.to_dict('records')
processed_records = []
for record in records:
    processed = {
        "Address": record["Address"],
        "city": record["Zip/city"],  # 匹配模型中的city字段
        "Price": record["Price"],
        "date": record["Date"]  # 匹配模型中的date字段
    }
    processed_records.append(processed)

# 批量创建对象并插入
MD = [MarketData(**data) for data in processed_records]
MarketData.objects.insert(MD, load_bulk=False)

方法二：修复JSON转换后的遍历逻辑 #

如果坚持使用JSON转换，需遍历json_object的值而非格式化后的字符串：

result = df.to_json(orient="index")
json_object = json.loads(result)

# 遍历json_object的values()，每个value是一条数据字典
MD = [MarketData(**{
    "Address": item["Address"],
    "city": item["Zip/city"],
    "Price": item["Price"],
    "date": item["Date"]
}) for item in json_object.values()]

MarketData.objects.insert(MD, load_bulk=False)

额外优化：模型字段规范 #

建议将模型字段改为更匹配数据类型的字段（而非全部用DynamicField），例如：

from mongoengine import StringField, DateTimeField, FloatField

class MarketData(Document):
    Address = StringField(required=False)
    city = StringField(required=False)
    Price = FloatField(required=False)  # 可先处理Price字符串转数值
    date = DateTimeField(required=False)
    
    # 可选：移除无用的json方法（或修正字段名错误）
    def json(self):
        market_dict = {
            "Address": self.Address,
            "city": self.city,
            "Price": self.Price,
            "date": self.date.isoformat() if self.date else None
        }
        return json.dumps(market_dict)

批量插入的性能优化 #

对于大量数据，推荐使用insert_many（MongoEngine支持）：

# 使用insert_many更高效
MarketData.objects.insert_many(processed_records, load_bulk=False)

总结 #

无需转回PyMongo，MongoEngine完全支持批量导入。核心问题是错误地遍历了JSON字符串而非字典对象，同时需注意DataFrame字段与模型字段的映射匹配。直接从DataFrame生成字典列表是最简洁高效的方式。

内容的提问来源于stack exchange，提问作者jmChrist