插入json时出现pymongo.errors.BulkWriteError，是什么原因造成的？

Question

由于我试图插入到数据库中的 jsons 没有正确格式化为 json，我将它附加到一个列表并 json.loads 它，最后将它全部插入到一个 mongodb 文档中。

我做错了什么？

当我尝试将其插入数据库时​​，出现以下错误：

Traceback (most recent call last):
  File "/media/anon/06bcf743-8b4d-409f-addc-520fc4e19299/PycharmProjects/LiveMe/import_messages_dev.py", line 41, in <module>
    messages.insert_many(json_data)
  File "/media/anon/06bcf743-8b4d-409f-addc-520fc4e19299/PycharmProjects/LiveMe/venv1/lib/python3.6/site-packages/pymongo/collection.py", line 742, in insert_many
    blk.execute(self.write_concern.document, session=session)
  File "/media/anon/06bcf743-8b4d-409f-addc-520fc4e19299/PycharmProjects/LiveMe/venv1/lib/python3.6/site-packages/pymongo/bulk.py", line 432, in execute
    return self.execute_command(generator, write_concern, session)
  File "/media/anon/06bcf743-8b4d-409f-addc-520fc4e19299/PycharmProjects/LiveMe/venv1/lib/python3.6/site-packages/pymongo/bulk.py", line 329, in execute_command
    raise BulkWriteError(full_result)
pymongo.errors.BulkWriteError: batch op errors occurred

我的代码如下：

###  DATABASE ####
# Connect to database // login user:password
uri = 'mongodb://testuser:password@ds245687.mlab.com:45687/liveme'
# Set client.
client = pymongo.MongoClient(uri)
# Set database.
db = client.get_database()

# Create collection.
messages = db['messages']

# The url to the live.me replays.
replay_url = "http://live.ksmobile.net/live/getreplayvideos?"

userid = 895324164037541888

# Parsing the urls for replays and profile with the userid.
url2 = replay_url + urllib.parse.urlencode({'userid': userid}) + '&page_size=1000'

# Printing urls for own validation.
print(f"Replay url: {url2}\n")

raw_replay_data = requests.get(url2).json()

# Insert messages to database.
for i in raw_replay_data['data']['video_info']:
    url3 = i['msgfile']
    raw_message_data = urllib.request.urlopen(url3)
    json_data = []
    for line in raw_message_data:
        json_data.append(json.loads(line))
        messages.insert_many(json_data)
        print(json_data)

更新以获取更多回答信息

Traceback (most recent call last):
  File "/media/anon/06bcf743-8b4d-409f-addc-520fc4e19299/PycharmProjects/LiveMe/import_messages_dev.py", line 51, in <module>
    msgfiles = [json.loads(line) for line in msgfile.iter_lines()]
  File "/media/anon/06bcf743-8b4d-409f-addc-520fc4e19299/PycharmProjects/LiveMe/import_messages_dev.py", line 51, in <listcomp>
    msgfiles = [json.loads(line) for line in msgfile.iter_lines()]
  File "/usr/lib/python3.6/json/__init__.py", line 354, in loads
    return _default_decoder.decode(s)
  File "/usr/lib/python3.6/json/decoder.py", line 339, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "/usr/lib/python3.6/json/decoder.py", line 357, in raw_decode
    raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

**第二次更新获取更多信息回答**

问题是，它仍然在一个文档中插入一行。 在前。 http://live.ksmobile.net/live/getreplayvideos?userid=895324164037541888&page_size=1000 每个“msgfile”包含 120 个“重播”。

我希望所有内容都是 1 "msgfile" == 1 mongodb 文档。 因此，遍历上述链接中的所有“msgfile”，并将 1 个“msgfile”中的所有内容插入到 1 个文档中。

"msgfile": "http://s.live.ksmobile.net/cheetahlive/86/9e/15207987261090823831/15207987261090823831.json"

"msgfile": "http://s.live.ksmobile.net/cheetahlive/ae/41/15206244909238491194/15206244909238491194.json",

等等..

这 120 个“msgfile”现在逐行插入会生成 65.113 个文档而不是 120 个文档。

Answer 1

第二个循环至少一直在构建，同时将每个构建插入到消息集合中。

建立要插入的消息列表，然后插入它们。 例如

示例 1。

...
for info in raw_replay_data['data']['video_info']:
    msgfile_url = info['msgfile']
    msgfile = requests.get(msgfile_url)
    msgfiles = [json.loads(line) for line in msgfile.iter_lines() if line.strip()]

    messages.insert_many(msgfiles)

示例 2。

您也可以建立msgfiles，然后将其插入到循环的末尾。

msgfiles = []
for info in raw_replay_data['data']['video_info']:
    msgfile_url = info['msgfile']
    msgfile = requests.get(msgfile_url)
    msgfiles.extend([json.loads(line) for line in msgfile.iter_lines() if line.strip()])
message.insert_many(msgfiles)

示例 3。

当您在插入大量数据时遇到 MongoDb 的错误时，您可以线程化您的请求。 例如

import json
from queue import Queue
from threading import Thread
...
q = Queue()
...

def consumer():
    while True:
        msgfiles = q.get()
        messages.insert_many(msgfiles)
        q.task_done()

...
for _ in range(6):
    t = Thread(target=consumer)
    t.start()

for info in raw_replay_data['data']['video_info']:
    msgfile_url = info['msgfile']
    msgfile = requests.get(msgfile_url)
    msgfiles = [json.loads(line) for line in msgfile.iter_lines() if line.strip()]

    q.put(msgfiles)

q.join()

编辑

在视频信息 URL 中插入消息文件作为单个文档。

示例 1。

for info in raw_replay_data['data']['video_info']:
    msgfile_url = info['msgfile']
    msgfile = requests.get(msgfile_url)
    msgfiles = [json.loads(line) for line in msgfile.iter_lines() if line.strip()]
    document = {'url': msgfile_url, 'data': msgfiles}
    messages.insert_one(document)

示例 2。

msgfiles = []
for info in raw_replay_data['data']['video_info']:
    msgfile_url = info['msgfile']
    msgfile = requests.get(msgfile_url)
    document = {
        'url': msgfile_url,
        'data': [json.loads(line) for line in msgfile.iter_lines() if line.strip()]
    }
    msgfiles.append(document)
message.insert_many(msgfiles)