MongoDB spostare i documenti da una raccolta a un'altra raccolta

Question

Come può documenti essere spostati da una raccolta a un'altra raccolta in MongoDB ?? Ad esempio: ho molti documenti nella raccolta A e voglio spostare tutti i documenti più vecchi di 1 mese alla raccolta B (questi documenti più vecchi di 1 mese non dovrebbero essere nella raccolta A).

Utilizzo dell'aggregazione copia. Ma quello che sto cercando di fare è spostando di documenti. Quale metodo può essere utilizzato per spostare i documenti?

Answer 1

Aggiornamento

This answer by @jasongarber è un approccio più sicuro e dovrebbe essere usato al posto mio.

---

Purché ti ho preso a destra e si desidera spostare tutti i documenti più vecchi di 1 mese, e si utilizza MongoDB 2.6, non v'è alcuna ragione per non utilizzare le operazioni di massa, che sono il modo più efficiente di fare più operazioni sono a conoscenza di:

> var bulkInsert = db.target.initializeUnorderedBulkOp() 
> var bulkRemove = db.source.initializeUnorderedBulkOp() 
> var date = new Date() 
> date.setMonth(date.getMonth() -1) 
> db.source.find({"yourDateField":{$lt: date}}).forEach(
    function(doc){ 
     bulkInsert.insert(doc); 
     bulkRemove.find({_id:doc._id}).removeOne(); 
    } 
) 
> bulkInsert.execute() 
> bulkRemove.execute() 

Questo dovrebbe essere piuttosto veloce e ha il vantaggio che nel caso in cui qualcosa va storto durante l'inserimento di massa, i dati originali esiste ancora.

---

Modifica

Al fine di evitare troppa memoria per essere utilizzato, è possibile eseguire l'operazione di massa su ogni x documenti elaborati:

> var bulkInsert = db.target.initializeUnorderedBulkOp() 
> var bulkRemove = db.source.initializeUnorderedBulkOp() 
> var x = 10000 
> var counter = 0 
> var date = new Date() 
> date.setMonth(date.getMonth() -1) 
> db.source.find({"yourDateField":{$lt: date}}).forEach(
    function(doc){ 
     bulkInsert.insert(doc); 
     bulkRemove.find({_id:doc._id}).removeOne(); 
     counter ++ 
     if(counter % x == 0){ 
     bulkInsert.execute() 
     bulkRemove.execute() 
     bulkInsert = db.target.initializeUnorderedBulkOp() 
     bulkRemove = db.source.initializeUnorderedBulkOp() 
     } 
    } 
) 
> bulkInsert.execute() 
> bulkRemove.execute() 

Answer 2

è possibile utilizzare query di intervallo per ottenere dati da sourceCollection e conservare i dati del cursore nel ciclo variabile e su di esso e inserire a obiettivo di raccolta:

var doc = db.sourceCollection.find({ 
     "Timestamp":{ 
       $gte:ISODate("2014-09-01T00:00:00Z"), 
       $lt:ISODate("2014-10-01T00:00:00Z") 
     } 
}); 

doc.forEach(function(doc){ 
    db.targetCollection.insert(doc); 
}) 

Speranza in modo che aiuta !!

Answer 3

Inserire e rimuovere:

var documentsToMove = db.collectionA.find({}); 
documentsToMove.forEach(function(doc) { 
    db.collectionB.insert(doc); 
    db.collectionA.remove(doc); 
} 

Answer 4

Può essere dal punto di vista delle prestazioni è meglio rimuovere un sacco di documenti utilizzando un comando (soprattutto se si hanno gli indici per parte query) anziché eliminarne one by-one.

Ad esempio:

db.source.find({$gte: start, $lt: end}).forEach(function(doc){ 
    db.target.insert(doc); 
}); 
db.source.remove({$gte: start, $lt: end}); 

Answer 5

Questa è una riaffermazione della @ Markus W Mahlberg

Restituzione del favore - come funzione

function moveDocuments(sourceCollection,targetCollection,filter) { 
    var bulkInsert = targetCollection.initializeUnorderedBulkOp(); 
    var bulkRemove = sourceCollection.initializeUnorderedBulkOp(); 
    sourceCollection.find(filter) 
     .forEach(function(doc) { 
     bulkInsert.insert(doc); 
     bulkRemove.find({_id:doc._id}).removeOne(); 
     } 
) 
    bulkInsert.execute(); 
    bulkRemove.execute(); 
} 

Un esempio utilizzare

var x = {dsid:{$exists: true}}; 
moveDocuments(db.pictures,db.artifacts,x) 

per spostare tutti i documenti che hanno alto elemento di livello DSID dalle immagini alla raccolta manufatti

Answer 6

Da MongoDB 3.0, è possibile utilizzare il comando copyTo con la seguente sintassi:

db.source_collection.copyTo("target_collection") 

Quindi è possibile utilizzare il comando drop per rem Ove la vecchia collezione:

db.source_collection.drop() 

Answer 7

mi piace la risposta da @ markus-w-Mahlberg, però, a volte, ho visto la necessità di mantenere un po 'più semplice per le persone. Come tale ho un paio di funzioni che sono sotto. Si potrebbe naturalmente avvolgere la cosa qui con operatori di massa come ha fatto lui, ma questo codice funziona allo stesso modo con i vecchi e nuovi sistemi Mongo.

function parseNS(ns){ 
    //Expects we are forcing people to not violate the rules and not doing "foodb.foocollection.month.day.year" if they do they need to use an array. 
    if (ns instanceof Array){ 
     database = ns[0]; 
     collection = ns[1]; 
    } 
    else{ 
     tNS = ns.split("."); 
     if (tNS.length > 2){ 
      print('ERROR: NS had more than 1 period in it, please pass as an [ "dbname","coll.name.with.dots"] !'); 
      return false; 
     } 
     database = tNS[0]; 
     collection = tNS[1]; 
    } 
    return {database: database,collection: collection}; 
} 

function insertFromCollection(sourceNS, destNS, query, batchSize, pauseMS){ 
    //Parse and check namespaces 
    srcNS = parseNS(sourceNS); 
    destNS = parseNS(destNS); 
    if (srcNS == false || destNS == false){return false;} 

    batchBucket = new Array(); 
    totalToProcess = db.getDB(srcNS.database).getCollection(srcNS.collection).find(query,{_id:1}).count(); 
    currentCount = 0; 
    print("Processed "+currentCount+"/"+totalToProcess+"..."); 
    db.getDB(srcNS.database).getCollection(srcNS.collection).find(query).addOption(DBQuery.Option.noTimeout).forEach(function(doc){ 
     batchBucket.push(doc); 
     if (batchBucket.length > batchSize){ 
      db.getDB(destNS.database).getCollection(destNS.collection)insert(batchBucket); 
      currentCount += batchBucket.length; 
      batchBucket = []; 
      sleep (pauseMS); 
      print("Processed "+currentCount+"/"+totalToProcess+"...");  
     } 
    } 
    print("Completed"); 
} 

/** Example Usage: 
     insertFromCollection("foo.bar","foo2.bar",{"type":"archive"},1000,20);  

Si potrebbe, ovviamente, aggiungere un db.getSiblingDB(srcNS.database).getCollection(srcNS.collection).remove(query,true) Se si voleva rimuovere anche i record dopo che sono stati copiati nella nuova posizione. Il codice può essere facilmente costruito in questo modo per renderlo riavviabile.

Answer 8

$ out è utilizzare per creare la nuova collezione con i dati, in modo da utilizzare $ fuori

db.oldCollection.aggregate([{$out : "newCollection"}]) 

quindi utilizzare cadere

db.oldCollection.drop() 

Answer 9

ho pensato di arhieve 1000 record alla volta utilizzando bulkinsert e bulkdelete metodi di pymongo.

Per entrambi fonte e la destinazione

1. creare oggetti mongodb per la connessione al database.

2. istanziare gli oggetti di grandi dimensioni. Nota: ho creato anche un backup di oggetti bulk. Questo mi aiuterà a ripristinare l'inserimento o la rimozione quando si verifica un errore. esempio:

   Per la sorgente // replace this with mongodb object creation logic source_db_obj = db_help.create_db_obj(source_db, source_col) source_bulk = source_db_obj.initialize_ordered_bulk_op() source_bulk_bak = source_db_obj.initialize_ordered_bulk_op()
   Per obiettivo // replace this with mogodb object creation logic target_db_obj = db_help.create_db_obj(target_db, target_col) target_bulk = target_db_obj.initialize_ordered_bulk_op() target_bulk_bak = target_db_obj.initialize_ordered_bulk_op()

3. Ottenere i record di origine che corrisponde ai criteri di filtro

   source_find_results = source_db_obj.find (filtro)

4. Loop attraverso la fonte registra

   Crea target e sfusi fonte operazioni

   accodamento archived_at campo con il datetime corrente alla raccolta di destinazione

   //replace this with the logic to obtain the UTCtime. doc['archived_at'] = db_help.getUTCTime() target_bulk.insert(document) source_bulk.remove(document)

   per ripristino in caso di errori o eccezioni, creare target_bulk_bak e operazioni source_bulk_bak.

   target_bulk_bak.find({'_id':doc['_id']}).remove_one() source_bulk_bak.insert(doc) //remove the extra column doc.pop('archieved_at', None)

5. Quando il conteggio dei record a 1000, eseguire il bersaglio - inserimento e fonte bulk - rimozione bulk. Nota: questo metodo richiede oggetti target_bulk e source_bulk per l'esecuzione.

   execute_bulk_insert_remove (source_bulk, target_bulk)

6. Quando si verifica un'eccezione, eseguire la rimozione target_bulk_bak e inesertions source_bulk_bak. Ciò ripristinerebbe le modifiche. Dal momento che MongoDB non ha rollback, sono arrivato fino a questo hack

   execute_bulk_insert_remove (source_bulk_bak, target_bulk_bak)

7. Infine reinizializzare gli oggetti di origine e di destinazione e alla rinfusa bulk_bak. Questo è necessario perché puoi usarli solo una volta.

8. completo del codice

   def execute_bulk_insert_remove(source_bulk, target_bulk): 
        try: 
         target_bulk.execute() 
         source_bulk.execute() 
        except BulkWriteError as bwe: 
         raise Exception(
          "could not archive document, reason: {}".format(bwe.details)) 
   
       def archive_bulk_immediate(filter, source_db, source_col, target_db, target_col): 
        """ 
        filter: filter criteria for backup 
        source_db: source database name 
        source_col: source collection name 
        target_db: target database name 
        target_col: target collection name 
        """ 
        count = 0 
        bulk_count = 1000 
   
        source_db_obj = db_help.create_db_obj(source_db, source_col) 
        source_bulk = source_db_obj.initialize_ordered_bulk_op() 
        source_bulk_bak = source_db_obj.initialize_ordered_bulk_op() 
   
        target_db_obj = db_help.create_db_obj(target_db, target_col) 
        target_bulk = target_db_obj.initialize_ordered_bulk_op() 
        target_bulk_bak = target_db_obj.initialize_ordered_bulk_op() 
   
        source_find_results = source_db_obj.find(filter) 
   
        start = datetime.now() 
   
        for doc in source_find_results: 
         doc['archived_at'] = db_help.getUTCTime() 
   
         target_bulk.insert(doc) 
         source_bulk.find({'_id': doc['_id']}).remove_one() 
         target_bulk_bak.find({'_id': doc['_id']}).remove_one() 
         doc.pop('archieved_at', None) 
         source_bulk_bak.insert(doc) 
   
         count += 1 
   
         if count % 1000 == 0: 
          logger.info("count: {}".format(count)) 
          try: 
           execute_bulk_insert_remove(source_bulk, target_bulk) 
          except BulkWriteError as bwe: 
           execute_bulk_insert_remove(source_bulk_bak, target_bulk_bak) 
           logger.info("Bulk Write Error: {}".format(bwe.details)) 
           raise 
   
          source_bulk = source_db_obj.initialize_ordered_bulk_op() 
          source_bulk_bak = source_db_obj.initialize_ordered_bulk_op() 
   
          target_bulk = target_db_obj.initialize_ordered_bulk_op() 
          target_bulk_bak = target_db_obj.initialize_ordered_bulk_op() 
   
        end = datetime.now() 
   
        logger.info("archived {} documents to {} in ms.".format(
         count, target_col, (end - start))) 
   

Answer 10

Le operazioni di massa @ Markus-w-Mahlberg ha mostrato (e @ mark-Mullin raffinato) sono efficienti, ma non sicuro come scritto. Se bulkInsert ha esito negativo, bulkRemove continuerà comunque. Per essere sicuri di non perdere tutti i record quando in movimento, utilizzare questo invece:

function insertBatch(collection, documents) { 
 
    var bulkInsert = collection.initializeUnorderedBulkOp(); 
 
    var insertedIds = []; 
 
    var id; 
 
    documents.forEach(function(doc) { 
 
    id = doc._id; 
 
    // Insert without raising an error for duplicates 
 
    bulkInsert.find({_id: id}).upsert().replaceOne(doc); 
 
    insertedIds.push(id); 
 
    }); 
 
    bulkInsert.execute(); 
 
    return insertedIds; 
 
} 
 

 
function deleteBatch(collection, documents) { 
 
    var bulkRemove = collection.initializeUnorderedBulkOp(); 
 
    documents.forEach(function(doc) { 
 
    bulkRemove.find({_id: doc._id}).removeOne(); 
 
    }); 
 
    bulkRemove.execute(); 
 
} 
 

 
function moveDocuments(sourceCollection, targetCollection, filter, batchSize) { 
 
    print("Moving " + sourceCollection.find(filter).count() + " documents from " + sourceCollection + " to " + targetCollection); 
 
    var count; 
 
    while ((count = sourceCollection.find(filter).count()) > 0) { 
 
    print(count + " documents remaining"); 
 
    sourceDocs = sourceCollection.find(filter).limit(batchSize); 
 
    idsOfCopiedDocs = insertBatch(targetCollection, sourceDocs); 
 

 
    targetDocs = targetCollection.find({_id: {$in: idsOfCopiedDocs}}); 
 
    deleteBatch(sourceCollection, targetDocs); 
 
    } 
 
    print("Done!") 
 
}