Проблемы с памятью на knex.js при использовании потоков

Question

Я пытаюсь экспортировать всю таблицу базы данных sqlite3 в CSV, используя knex.js. Поскольку таблица может содержать до 300000 строк, я использую потоки, чтобы не иметь проблем с памятью. Но если я смотрю на использование памяти в моем приложении, это до 800 МБ или у меня ошибка «из памяти».

Как я могу обработать большой результат запроса с помощью knex.js в базе данных sqlite3?

Ниже образец кода:

knex.select().from(table).stream(function (stream) { 
    var stringifier = stringify(opts); 
    var fileStream = fs.createWriteStream(file); 

    var i = 0; 
    stringifier.on('readable', function() { 
     var row; 
     while (row = stringifier.read()) { 
     fileStream.write(row); 
     console.log("row " + i++); //debug 
     } 
    }); 

    fileStream.once('open', function(fd) { 
     stream.pipe(stringifier); 
    }); 
}); 

EDIT

Кажется knex.js потоки для sqlite3 базы данных являются "поддельные" потоки. Ниже исходный код функции потока для sqlite3 в knex:

Runner_SQLite3.prototype._stream = Promise.method(function(sql, stream, options) { 
    /*jshint unused: false*/ 
    var runner = this; 
    return new Promise(function(resolver, rejecter) { 
     stream.on('error', rejecter); 
     stream.on('end', resolver); 
     return runner.query(sql).map(function(row) { 
      stream.write(row); 
     }).catch(function(err) { 
      stream.emit('error', err); 
     }).then(function() { 
      stream.end(); 
     }); 
    }); 
}); 

Мы видим, что он ожидает запрос будет выполняться до создания потока из массива результатов.

версия:

• Knex.Js 0.7.5
• узел 0,12

Thx за вашу помощь.

Answer 1

Я думаю, что нет никаких решений. Я использую лимит и смещение, чтобы все данные были шаг за шагом с knex.js и я записывал каждую строку в поток записи. Пример внедрения для тех, кто хочет:

exportTable: function(table, writeStream) { 
    var totalRows; 
    var rowLimit = _config.ROW_LIMIT; 

    return DatabaseManager.countAll(table).then(function(count) { 

     totalRows = count[0]['count(*)']; 
     var iterations = new Array(Math.ceil(totalRows/rowLimit)); 

     return Promise.reduce(iterations, function(total, item, index) { 

      return _knex.select().from(table).limit(rowLimit).offset(index * rowLimit).map(function(row) { 
       writeStream.write(row); 
      }).catch(function(err) { 
       return Promise.reject(err); 
      }); 

     }, 0).then(function() { 
      return Promise.resolve(); 
     }).catch(function(err) { 
      return Promise.reject(err); 
     }); 

    }).catch(function(err) { 
     console.log(err); 
     return Promise.reject(err); 
    }); 
}