Я пишу программу NodeJS, которая проверяет целостность данных, предоставляемых в формате CSV (в виде буферов).Для этого я использую пакет csvdata npm, но так как пакет принимает только пути к файлам в качестве входных данных, мне пришлось сделать разветвление и изменить его, чтобы принимать в качестве входных данных буферы (для этогочтобы удовлетворить мои текущие потребности).
Чтобы выполнить обработку файла, пакет csvdata изначально выполняет следующее:
fs.createReadStream (filePath,{options}) -> pipe -> csv.parse () -> checkIntegrity
, и я пытаюсь сделать следующее:
Буферизованный файл -> ReadStream (буфер, {опции}) (это будет разбито на буферизованный файл) -> pipe -> csv.parse () -> checkIntegrity
Проблема, с которой я застрял, связана с производительностью.Например, при обработке файла CSV размером 283 МБ:
- (оригинальный путь - filePath): до 40 секунд.
- (с использованием буфера с ReadableStream): 450 секунд (более 7
минуты ...)
Я протестировал много пакетов для создания потока из файлового буфера, включая streamifier , stream-buffers , нони один из них не дал мне хороших результатов, поэтому я в конечном итоге использовал решение, которое нашел на этом сайте (https://www.bennadel.com/blog/2681-turning-buffers-into-readable-streams-in-node-js.htm), и изменил его в соответствии с моими потребностями (это дает мне 450 секунд, остальные хуже)..):
//Required module references.
var stream = require( "stream" );
var util = require( "util" );
// ---------------------------------------------------------- //
// ---------------------------------------------------------- //
// I turn the given source Buffer into a Readable stream.
function BufferStream( source ) {
if ( ! Buffer.isBuffer( source ) ) {
throw( new Error( "Source must be a buffer." ) );
}
// Super constructor.
stream.Readable.call( this, {encoding: 'utf-8', highWaterMark: 64 * 1024 } );
this._source = source;
// I keep track of which portion of the source buffer is currently being pushed
// onto the internal stream buffer during read actions.
this._offset = 0;
this._length = source.length;
// When the stream has ended, try to clean up the memory references.
this.on( "end", this._destroy );
}
util.inherits( BufferStream, stream.Readable );
// I attempt to clean up variable references once the stream has been ended.
// --
// NOTE: I am not sure this is necessary. But, I'm trying to be more cognizant of memory
// usage since my Node.js apps will (eventually) never restart.
BufferStream.prototype._destroy = function() {
this._source = null;
this._offset = null;
this._length = null;
};
// I read chunks from the source buffer into the underlying stream buffer.
// --
// NOTE: We can assume the size value will always be available since we are not
// altering the readable state options when initializing the Readable stream.
BufferStream.prototype._read = function( size ) {
// If we haven't reached the end of the source buffer, push the next chunk onto
// the internal stream buffer.
if ( this._offset < this._length ) {
this.push( this._source.slice( this._offset, ( this._offset + size ) ) );
this._offset += size;
// If we've consumed the entire source buffer, close the readable stream.
// if ( this._offset >= this._length ) {
} else {
this.push( null );
}
};
// I finally create the BufferStream
var dataStream = new BufferStream(buffer)
// console.log(firstStream)
return dataStream.pipe(csv.parse(parseOpts))
Наконец, я использовал console.log, чтобы проверить разницу между обоими потоками (один, созданный с помощью fs.createReadStream, и один, созданный с помощью приведенного выше кода) и не вижу какой-либо существенной разницыэто объясняет разницу в производительности.
Случай 1: это console.log из потока, созданного с помощью fs.createReadStream ()
ReadStream {
_readableState:
ReadableState {
objectMode: false,
highWaterMark: 65536,
buffer: BufferList { head: null, tail: null, length: 0 },
length: 0,
pipes: null,
pipesCount: 0,
flowing: null,
ended: false,
endEmitted: false,
reading: false,
sync: true,
needReadable: false,
emittedReadable: false,
readableListening: false,
resumeScheduled: false,
paused: true,
emitClose: false,
autoDestroy: false,
destroyed: false,
defaultEncoding: 'utf8',
awaitDrain: 0,
readingMore: false,
decoder:
StringDecoder {
encoding: 'utf8',
[Symbol(kNativeDecoder)]: <Buffer 00 00 00 00 00 00 01> },
encoding: 'utf8' },
readable: true,
_events: [Object: null prototype] { end: [Function] },
_eventsCount: 1,
_maxListeners: undefined,
path: ‘./data.csv',
fd: null,
flags: 'r',
mode: 438,
start: undefined,
end: Infinity,
autoClose: true,
pos: undefined,
bytesRead: 0,
closed: false }
И это результат console.logсделал по этому обычаю "BufferStream"
BufferStream {
_readableState:
ReadableState {
objectMode: false,
highWaterMark: 65536,
buffer: BufferList { head: null, tail: null, length: 0 },
length: 0,
pipes: null,
pipesCount: 0,
flowing: null,
ended: false,
endEmitted: false,
reading: false,
sync: true,
needReadable: false,
emittedReadable: false,
readableListening: false,
resumeScheduled: false,
paused: true,
emitClose: true,
autoDestroy: false,
destroyed: false,
defaultEncoding: 'utf8',
awaitDrain: 0,
readingMore: false,
decoder:
StringDecoder {
encoding: 'utf8',
[Symbol(kNativeDecoder)]: <Buffer 00 00 00 00 00 00 01> },
encoding: 'utf-8' },
readable: true,
_events: [Object: null prototype] { end: [Function] },
_eventsCount: 1,
_maxListeners: undefined,
_source:
<Buffer ef bb bf 46 65 63 68 61 46 61 63 74 75 72 61 2c 50 61 69 73 2c 4e 6f 6d 50 61 69 73 2c 53 4b 55 2c 4b 69 6c 6f 73 46 61 63 74 75 72 61 64 6f 73 2c 4b ... 282660848 more bytes>,
_offset: 0,
_length: 282660898 }
Извините за длинное объяснение, но я думаю, что это вся информация, которую я могу предоставить, которая могла бы помочь.Я искал в Интернете и в документах, пытаясь понять, что может объяснить разницу, но не смог этого сделать ... Может быть, именно так fs.createReadStream реализован на низком уровне.Я действительно не знаю ... потому что для меня нелогично думать, что файл, который читается с диска, а затем обрабатывается в потоке, может быть быстрее буфера (который уже находится в памяти, так как я начинаю отсчитывать время послефайл загружен), который преобразуется в поток для последующей обработки.Так что моя интуиция подсказывает мне, что я либо делаю ошибку нуби, либо не понимаю что-то о потоках, либо он работает так, как работает fs.createReadStream ... Любая помощь или руководство будут высоко оценены !!