--- /dev/null
+// ## Example 7: Processing rows in batches (multi-row input)
+//
+// Set the `maxBatchSize` option to a number greater than `1` to collect rows into batches before passing the whole
+// batch to the transformation function.
+//
+// This functionality is useful if you need to collect data from multiple rows before processing (e.g., performing an
+// expensive network request where is more efficient to batch multiple rows into one request).
+
+import { createCSVReadableStream, createCSVTransformStream, createCSVWritableStream } from '../src/index.js';
+
+let firstChunk = true;
+let batchNumber = 0;
+function process(batch) {
+ if (firstChunk) { // Header row is always passed by itself, regardless of maxBatchSize
+ firstChunk = false;
+ return ['batch number', 'sum of batch', ...batch]; // Note the header row is Array<string>, NOT Array<Array<string>>
+ }
+
+ // Some logic that requires data across multiple rows
+ let sum = 0;
+ for (const row of batch) {
+ for (const field of row) {
+ sum += Number(field);
+ }
+ }
+
+ batchNumber++;
+ return batch.map(row => [batchNumber, sum, ...row]);
+}
+
+await createCSVReadableStream(new URL('./data/ex7-in.csv', import.meta.url))
+ .pipeThrough(createCSVTransformStream(process, { includeHeaders: true, maxBatchSize: 5 }))
+ .pipeTo(createCSVWritableStream(new URL('./data/ex7-out.csv', import.meta.url)));
\ No newline at end of file
}
/**
- * A row of input CSV data.
- * @typedef {Array<string>|string} TransformationInput
+ * A row or rows of input CSV data.
+ * @typedef {Array<Array<string>>|Array<string>|string} TransformationInput
*/
/**
*/
/**
- * A function to process a row of CSV data from `createCSVReadableStream`.
+ * A function to process a row or rows of CSV data from `createCSVReadableStream`.
* @callback TransformationFunction
* @param {TransformationInput} row
* @returns {TransformationOutput}
* @property {null|function(TransformationInput,Error,TransformationFunction):TransformationOutput} [onError=null] Set to a function to catch
* errors thrown by the transformation function. The input data, the error that was thrown, and the transformation function itself will be passed
* to the `onError` function. The default value is `null` (errors will not be caught).
+ * @property {Number} [maxBatchSize=1] The maximum number of rows that will be passed to the transformation function per function call (greedy).
+ * The default value is `1`.
*/
/**
* Create a new [TransformStream](https://developer.mozilla.org/en-US/docs/Web/API/TransformStream) to process CSV data.
*
- * @param {TransformationFunction} fn A function to process a row of CSV data from `createCSVReadableStream`.
+ * @param {TransformationFunction} fn A function to process a row or rows of CSV data from `createCSVReadableStream`.
*
- * If `options.rawInput` is `false` (default), the input will be a `Array<string>` representing the CSV row. If `options.rawInput` is `true`,
- * the input will be a JSON `string` representing the CSV row.
+ * If `options.rawInput` is `false` (default), the input will be a `Array<string>` representing a CSV row. If `options.rawInput` is `true`,
+ * the input will be a JSON `string` representing a CSV row.
*
* If `options.rawOutput` is `false` (default), the expected output is a `Array<any>` which will be serialized with `JSON.stringify()` and enqueued,
* or a `Array<Array<any>>` which will have each of its sub-arrays serialized and enqueued separately. If `options.rawOutput` is `true`, the raw
* output will be sent the next stream unmodified.
*
+ * If `options.maxBatchSize` is a number greater than `1`, the input will represent multiple CSV rows as an `Array<Array<string>>` (if
+ * `options.rawInput` is `false`) or a `string` (if `options.rawInput` is `true`). The input will still be a 2-D array even if there is only
+ * 1 row in the batch (except for the header row). The header row is always processed by itself and passed as a `Array<string>`, regardless of
+ * the `options.maxBatchSize` value.
+ *
* Return `null` to consume an input row without emitting an output row.
* @param {CreateCSVTransformStreamOptions} options Object containing flags to configure the stream logic.
* @returns {TransformStream} A `TransformStream` where each chunk is one row of the CSV file, after transformations applied by `fn()`.
options.rawInput ??= false;
options.rawOutput ??= false;
options.onError ??= null;
+ options.maxBatchSize ??= 1;
async function wrappedFn(row) {
if (options.onError === null) { return await fn(row); }
catch (e) { return await options.onError(row, e, fn); }
}
+ const batch = [];
let firstChunk = true;
return new TransformStream({
async transform(chunk, controller) {
firstChunk = false;
out = options.includeHeaders ? await wrappedFn(row) : row;
}
+ else if (options.maxBatchSize > 1) {
+ batch.push(row);
+ if (batch.length === options.maxBatchSize) {
+ out = await wrappedFn(batch);
+ batch.length = 0;
+ }
+ }
else {
out = await wrappedFn(row);
}
else {
controller.enqueue(JSON.stringify(out));
}
+ },
+ async flush(controller) {
+ if (batch.length > 0) { // CSV ended before batch got filled
+ const out = await wrappedFn(batch);
+ if (out === null || out === undefined) { // Input row is consumed without emitting any output row
+ return;
+ }
+ if (options.rawOutput || typeof out === 'string') {
+ controller.enqueue(out);
+ }
+ else if (Array.isArray(out) && Array.isArray(out[0])) { // Multiple rows returned, enqueue each row separately
+ out.forEach(row => controller.enqueue(JSON.stringify(row)));
+ }
+ else {
+ controller.enqueue(JSON.stringify(out));
+ }
+ }
}
});
}
{ name: 'Error', message: 'inner' }
);
});
+ test('creates one batch when maxBatchSize is greater than CSV row length', { concurrency: true }, async () => {
+ await createCSVMockStream([
+ ['columnA', 'columnB'],
+ ['1', '2'],
+ ['3', '4'],
+ ['5', '6']
+ ])
+ .pipeThrough(createCSVTransformStream(b => b.map(r => r.map(f => Number(f) * 2)), { maxBatchSize: 100 }))
+ .pipeTo(csvStreamEqualWritable([
+ ['columnA', 'columnB'],
+ [2, 4],
+ [6, 8],
+ [10, 12]
+ ]));
+ });
+ test('creates one batch when maxBatchSize is equal to CSV row length', { concurrency: true }, async () => {
+ await createCSVMockStream([
+ ['columnA', 'columnB'],
+ ['1', '2'],
+ ['3', '4'],
+ ['5', '6']
+ ])
+ .pipeThrough(createCSVTransformStream(b => b.map(r => r.map(f => Number(f) * 2)), { maxBatchSize: 3 }))
+ .pipeTo(csvStreamEqualWritable([
+ ['columnA', 'columnB'],
+ [2, 4],
+ [6, 8],
+ [10, 12]
+ ]));
+ });
+ test('creates multiple batches when maxBatchSize is less than CSV row length', { concurrency: true }, async () => {
+ await createCSVMockStream([
+ ['columnA', 'columnB'],
+ ['1', '2'],
+ ['3', '4'],
+ ['5', '6']
+ ])
+ .pipeThrough(createCSVTransformStream(b => b.map(r => r.map(f => Number(f) * 2)), { maxBatchSize: 2 }))
+ .pipeTo(csvStreamEqualWritable([
+ ['columnA', 'columnB'],
+ [2, 4],
+ [6, 8],
+ [10, 12]
+ ]));
+ });
});
suite('createCSVWritableStream', { concurrency: true }, () => {
git.ardis.lu / csvio / commitdiff