流插入XML數據的數據庫

Question

我試圖有效地插入大量的數據（XML文件超過70GB的大小）沒有崩潰我的MongoDB服務器。目前，這就是我在使用的NodeJS做xml-stream：

var fs = require('fs'), 
    path = require('path'), 
    XmlStream = require('xml-stream'), 
    MongoClient = require('mongodb').MongoClient, 
    assert = require('assert'), 
    ObjectId = require('mongodb').ObjectID, 
    url = 'mongodb://username:password@my.server:27017/mydatabase', 
    amount = 0; 

var stream = fs.createReadStream(path.join(__dirname, 'motor.xml')); 
var xml = new XmlStream(stream); 

xml.collect('ns:Statistik'); 
xml.on('endElement: ns:Statistik', function(item) { 
    var insertDocument = function(db, callback) { 
     db.collection('vehicles').insertOne(item, function(err, result) { 
      amount++; 
      if (amount % 1000 == 0) { 
       console.log("Inserted", amount); 
      } 
      callback(); 
     }); 
    }; 

    MongoClient.connect(url, function(err, db) { 
     insertDocument(db, function() { 
      db.close(); 
     }); 
    }); 
}); 

當我打電話xml.on()它基本上返回樹/元素，我目前。由於這是JSON的直線，所以我可以將它作爲參數提供給我的db.collection().insertOne()函數，並將其按照我的需要插入到數據庫中。

所有代碼的實際工作，因爲它是現在，但經過約3000插入停止（約需10秒）。我懷疑這是因爲我打開數據庫連接，插入數據，然後每次在XML文件中看到一棵樹時都關閉連接，在這種情況下，大約有3000次。

我可以在某種程度上將insertMany()函數合併爲100個（或更多）的塊，但我不太確定這將如何處理這一切，這些工作都是流式傳輸和異步處理。

所以我的問題是：如何插入大量XML（以JSON）到我的MongoDB數據庫沒有它崩潰？

Answer 1

你假設.insertMany()比寫每一次都好，所以它只是收集"stream"上的數據。

因爲執行是「異步」，你通常要避免在棧上太多的主動呼叫，所以通常你調用.insertMany()然後.resume()一旦回調完成之前.pause()的"stream"：

var fs = require('fs'), 
    path = require('path'), 
    XmlStream = require('xml-stream'), 
    MongoClient = require('mongodb').MongoClient, 
    url = 'mongodb://username:password@my.server:27017/mydatabase', 
    amount = 0; 

MongoClient.connect(url, function(err, db) { 

    var stream = fs.createReadStream(path.join(__dirname, 'motor.xml')); 
    var xml = new XmlStream(stream); 

    var docs = []; 
    //xml.collect('ns:Statistik'); 

    // This is your event for the element matches 
    xml.on('endElement: ns:Statistik', function(item) { 
     docs.push(item);   // collect to array for insertMany 
     amount++; 

     if (amount % 1000 === 0) { 
      xml.pause();    // pause the stream events 
      db.collection('vehicles').insertMany(docs, function(err, result) { 
      if (err) throw err; 
      docs = [];    // clear the array 
      xml.resume();   // resume the stream events 
      }); 
     } 
    }); 

    // End stream handler - insert remaining and close connection 
    xml.on("end",function() { 
     if (amount % 1000 !== 0) { 
     db.collection('vehicles').insertMany(docs, function(err, result) { 
      if (err) throw err; 
      db.close(); 
     }); 
     } else { 
     db.close(); 
     } 
    }); 

}); 

甚至現代化不是有點：

const fs = require('fs'), 
     path = require('path'), 
     XmlStream = require('xml-stream'), 
     MongoClient = require('mongodb').MongoClient; 

const uri = 'mongodb://username:password@my.server:27017/mydatabase'; 

(async function() { 

    let amount = 0, 
     docs = [], 
     db; 

    try { 

    db = await MongoClient.connect(uri); 

    const stream = fs.createReadStream(path.join(__dirname, 'motor.xml')), 
      xml = new XmlStream(stream); 

    await Promise((resolve,reject) => { 
     xml.on('endElement: ns:Statistik', async (item) => { 
     docs.push(item); 
     amount++; 

     if (amount % 1000 === 0) { 
      try { 
      xml.pause(); 
      await db.collection('vehicle').insertMany(docs); 
      docs = []; 
      xml.resume(); 
      } catch(e) { 
      reject(e) 
      } 
     } 

     }); 

     xml.on('end',resolve); 

     xml.on('error',reject); 
    }); 

    if (amount % 1000 !== 0) { 
     await db.collection('vehicle').insertMany(docs); 
    } 

    } catch(e) { 
    console.error(e); 
    } finally { 
    db.close(); 
    } 

})(); 

注意，MongoClient連接實際上封裝了所有的其他操作。你只需要連接一次，以及其他操作發生在對"stream"事件處理程序。

因此，對於你XMLStream的觸發事件處理程序的表達式匹配和數據提取和收集到一個數組。每調用1000個項目.insertMany()就會插入文檔，在「異步」調用上「暫停」和「恢復」。

一旦完成「結束」事件的"stream"解僱。這是關閉數據庫連接的地方，事件循環將被釋放並結束程序。

儘管可以通過允許各種各樣的.insertMany()調用一次發生（通常爲「合併大小」以避免溢出調用堆棧），從而獲得某種程度的「並行性」，但基本上這是該過程的外觀最簡單的形式就是在等待其他異步I/O完成時簡單地暫停。

注意：註釋從你的原代碼.collect()方法按照follow up question這似乎並不是必需的，實際上是保留真的應該每次使用後丟棄在存儲節點寫入數據庫。