Fast or Bulk Upsert en pymongo

Question

¿Cómo puedo hacer un postre en bulto en pymongo? Quiero actualizar un montón de entradas y hacerlas de una en una es muy lento.

La respuesta a una pregunta casi idéntica está aquí: Bulk update/upsert in MongoDB?

La respuesta aceptada en realidad no responde a la pregunta. Simplemente da un enlace a la CLI de mongo para hacer importaciones/exportaciones.

También estaría abierto a que alguien explique por qué no es posible realizar una restauración masiva/no es una buena práctica, pero explique cuál es la solución preferida para este tipo de problema.

Gracias!

Answer 1

La respuesta sigue siendo la misma: no se admite el envío masivo de mensajes.

Answer 2

Puede actualizar todos los documentos que coincidan con su especificación de consulta utilizando multi = True.

Hay un error here sobre hacer un lote de comandos de la manera que desee.

Answer 3

MongoDB 2.6+ tiene soporte para operaciones masivas. Esto incluye inserciones en bloque, actualizaciones, actualizaciones, etc. El objetivo de esto es reducir/eliminar las demoras de la latencia de ida y vuelta de las operaciones de registro por registro ('documento por documento' es correcto).

Entonces, ¿cómo funciona esto? Ejemplo en Python, porque eso es lo que estoy trabajando.

>>> import pymongo 
>>> pymongo.version 
'2.7rc0' 

Para utilizar esta función, se crea un objeto 'a granel', añadir documentos a la misma, a continuación, llamar ejecutar en él y se enviará todas las actualizaciones En seguida. Advertencias: El tamaño BSON de las operaciones recopiladas (suma de los bsonsizes) no puede sobrepasar el límite de tamaño de documento de 16 MB. Por supuesto, el número de operaciones puede variar significativamente, su millaje puede variar.

Ejemplo en Pymongo de operación upsert granel:

import pymongo 
conn = pymongo.MongoClient('myserver', 8839) 
db = conn['mydbname'] 
coll = db.myCollection 
bulkop = coll.initialize_ordered_bulk_op() 
retval = bulkop.find({'field1':1}).upsert().update({'$push':{'vals':1}) 
retval = bulkop.find({'field1':1}).upsert().update({'$push':{'vals':2}) 
retval = bulkop.find({'field1':1}).upsert().update({'$push':{'vals':3}) 
retval = bulkop.execute() 

Este es el método esencial. Más información disponible en:

http://api.mongodb.org/python/2.7rc1/examples/bulk.html

Answer 4

disponible sólo en pymongo 2.7+

http://api.mongodb.org/python/current/examples/bulk.html

Answer 5

lanzamientos modernos de pymongo (mayor que 3.x) envuelven las operaciones a granel en una interfaz consistente que rebaja en donde la versión del servidor no admite operaciones masivas. Esto ahora es consistente en los controladores oficialmente compatibles con MongoDB.

Por lo tanto, el método preferido para la codificación es usar bulk_write() en su lugar, donde utiliza una UpdateOne otra acción de operación apropiada en su lugar.Y ahora, por supuesto, se prefiere usar las listas de lenguaje natural en lugar de un constructor específico

La traducción directa de la antigua documention:

from pymongo import UpdateOne 

operations = [ 
    UpdateOne({ "field1": 1},{ "$push": { "vals": 1 } },upsert=True), 
    UpdateOne({ "field1": 1},{ "$push": { "vals": 2 } },upsert=True), 
    UpdateOne({ "field1": 1},{ "$push": { "vals": 3 } },upsert=True) 
] 

result = collection.bulk_write(operations) 

o el clásico circuito de transformación de documentos:

import random 
from pymongo import UpdateOne 

random.seed() 

operations = [] 

for doc in collection.find(): 
    # Set a random number on every document update 
    operations.append(
     UpdateOne({ "_id": doc["_id"] },{ "$set": { "random": random.randint(0,10) } }) 
    ) 

    # Send once every 1000 in batch 
    if (len(operations) == 1000): 
     collection.bulk_write(operations,ordered=False) 
     operations = [] 

if (len(operations) > 0): 
    collection.bulk_write(operations,ordered=False) 

El resultado devuelto es de BulkWriteResult que contendrá los contadores de documentos adaptados y actualizados, así como los valores devueltos _id para cualquier "upserts" que se producen.

hay un poco de un error sobre el tamaño de la matriz de las operaciones a granel. La solicitud real tal como se envió al servidor no puede superar el límite de BSON de 16 MB, ya que ese límite también se aplica a la "solicitud" enviada al servidor que también utiliza el formato BSON.

Sin embargo que no gobierna el tamaño de la matriz de solicitud que se puede construir, ya que sólo se enviarán las operaciones reales y se procesan en lotes de 1000 de todos modos. La única restricción real es que esas 1000 instrucciones de operación en sí mismas no crean realmente un documento BSON de más de 16 MB. Que de hecho es una orden bastante difícil.

El concepto general de los métodos a granel es "menos tráfico", como resultado de enviar cosas a la vez y sólo se trata de una respuesta del servidor. La reducción de esa sobrecarga adjunta a cada solicitud de actualización ahorra mucho tiempo.

Answer 6

actualización masiva más rápido con Python 3.5 +, motor y asyncio:

import asyncio 
import datetime 
import logging 
import random 
import time 

import motor.motor_asyncio 
import pymongo.errors 


async def execute_bulk(bulk): 
    try: 
     await bulk.execute() 
    except pymongo.errors.BulkWriteError as err: 
     logging.error(err.details) 


async def main(): 
    cnt = 0 
    bulk = db.initialize_unordered_bulk_op() 
    tasks = [] 
    async for document in db.find({}, {}, no_cursor_timeout=True): 
     cnt += 1 
     bulk.find({'_id': document['_id']}).update({'$set': {"random": random.randint(0,10)}}) 
     if not cnt % 1000: 
      task = asyncio.ensure_future(execute_bulk(bulk)) 
      tasks.append(task) 
      bulk = db.initialize_unordered_bulk_op() 
    if cnt % 1000: 
     task = asyncio.ensure_future(bulk.execute(bulk)) 
     tasks.append(task) 
    logging.info('%s processed', cnt) 
    await asyncio.gather(*tasks) 


logging.basicConfig(level='INFO')  
db = motor.motor_asyncio.AsyncIOMotorClient()['database']['collection'] 
start_time = time.time() 
loop = asyncio.get_event_loop() 
try: 
    loop.run_until_complete(main()) 
finally: 
    execution_time = time.time() - start_time 
    logging.info('Execution time: %s', datetime.timedelta(seconds=execution_time)) 

Answer 7

si tiene muchos datos y desea utilizar "_id" para juzgar si existen datos,

puede probar ...

import pymongo 
from pymongo import UpdateOne 
client = pymongo.MongoClient('localhost', 27017) 
db=client['sampleDB'] 

collectionInfo = db.sample 

#sample data 
datas=[ 
    {"_id":123456,"name":"aaa","N":1,"comment":"first sample","lat":22,"lng":33}, 
    {"_id":234567,"name":"aaa","N":1,"comment":"second sample","lat":22,"lng":33}, 
    {"_id":345678,"name":"aaa","N":1,"comment":"xxx sample","lat":22,"lng":33}, 
    {"_id":456789,"name":"aaa","N":1,"comment":"yyy sample","lat":22,"lng":33}, 
    {"_id":123456,"name":"aaaaaaaaaaaaaaaaaa","N":1,"comment":"zzz sample","lat":22,"lng":33}, 
    {"_id":11111111,"name":"aaa","N":1,"comment":"zzz sample","lat":22,"lng":33} 
] 

#you should split judge item and other data 
ids=[data.pop("_id") for data in datas] 

operations=[UpdateOne({"_id":idn},{'$set':data},upsert=True) for idn ,data in zip(ids,datas)] 

collectionInfo.bulk_write(operations) 

Mi Inglés es muy pobre, si no puede entender lo que digo, lo siento