Batch

This utility is a Transform stream. It operates in object mode accepting items and packing them into an array. As soon as the array is big enough (a configurable value) it is emitted as an output. All arrays are as big as specified but the last one can be smaller but never empty.

Introduction

Example:

const batch = require('stream-json/utils/batch.js');

const {streamArray} = require('stream-json/streamers/stream-array.js');
const {chain} = require('stream-chain');
const fs = require('fs');

const pipeline = chain([fs.createReadStream('sample.json'), streamArray.withParser(), batch({batchSize: 100})]);

// count all odd values from a huge array

let oddCounter = 0;
pipeline.on('data', data => {
  console.log('Batch size:', data.length);
  data.forEach(pair => {
    if (pair.value % 2) ++oddCounter;
  });
});
pipeline.on('end', () => console.log('Odd numbers:', oddCounter));

API

The module returns a factory function. batch() returns a composable function for use in chain(). batch.asStream() returns a Transform stream for .pipe() usage.

constructor(options)

options is an optional object described in detail in node.js' Stream documentation. Additionally, the following custom options are recognized:

• batchSize is a positive integer number, which defines how many items should be placed into an array before it is outputted. Default: 1000.

Static methods and properties

batch.batch(options)

Alias of the factory function.

batch.asStream(options)

Returns a Transform stream suitable for .pipe() usage. The returned stream has a _batchSize property attached.

const batch = require('stream-json/utils/batch.js');
// ...

const pipeline = chain([fs.createReadStream('sample.json'), streamArray.withParser(), batch({batchSize: 100})]);
// ...

batch.asWebStream(options)

Returns a {readable, writable} pair suitable for Web Streams. The pair has a _batchSize property attached for parity with asStream.

Web Streams

batch ships in two substrate-specific entries with the same factory shape:

• Node — stream-json/utils/batch.js. Has asStream (Node Duplex) and asWebStream (Web pair). Both attach _batchSize to the returned object.
• Web — stream-json/web/utils/batch.js. Has asWebStream only. Pulls in no Node-stream imports.

Both factories return the same flushable, so chain on either substrate auto-wraps it.

// Web
import {chain} from 'stream-chain/web';
import {parser} from 'stream-json/web/parser.js';
import {streamArray} from 'stream-json/web/streamers/stream-array.js';
import batch from 'stream-json/web/utils/batch.js';

const pipeline = chain([source, parser(), streamArray(), batch({batchSize: 100})]);
for await (const arr of pipeline.readable) console.log('Batch size:', arr.length);