batch-process

In the real world, most async operations against large data sets need to be batched and throttled.

The batchProcess function provides a simple, configurable utility for batching and throttling async operations.

batchProcess logs the progress of batch operations to console or to an injected logging dependency.

API Documentation • CHANGELOG

Installation

npm i @karmaniverous/batch-process

Usage

import { batchProcess } from '@karmaniverous/batch-process';

type Item = Record<string, unknown>; // Your data type.

// Say you have a function processBatch that processes a batch of items, for
// example writing them to a database. Maybe not every item is processed
// successfully, so the function returns a count of processed items and an
// array of unprocessed ones.
interface AsyncResult {
  processed: number;
  unprocessed: Item[];
}

const processBatch = async (items: Item[]): AsyncResult =>
  doSomethingAsync(items);

// Let's write a function that extracts any unprocessed items from an
// AsyncResult.
const unprocessedItemExtractor = ({ unprocessed }: AsyncResult) => unprocessed;

// Here's a bunch of items to process.
const items: Item[] = [
  ... // Your items here.
];

// Now you can asynchronously process a bunch of items in throttled sets of
// parallel batches!

// batchProcess will...
// - break items into batches of no more than `batchSize` items, and
// - process up to `throttle` batches in parallel, and
// - write a debug log message for each batch attempted, and
// - wait `delayIncrement` ms with exponential backoff to retry failed batches, and
// - throw an exception after `maxRetries` failed retries, and
// - return an array of AsyncResults generated during batch processing.
const results = await batchProcess(items, {
  processBatch,
  unprocessedItemExtractor,
  // see other options below
});

batchProcess Options

The options parameter has the following properties:

Default Batchable options are:

| Option | Type | Default | Description | | -------------------------- | ---------------------------------------- | ------------ | ------------------------------------------------------------------------------------------------------------------- | | batchHandler | (items: Item[]) => Promise<Output> | required | Async function to process an individual batch. | | batchSize | number | 25 | The number of items to process in each batch. | | delayIncrement | number | 100 | The number of milliseconds to wait before retrying a failed batch, with 2x exponential backoff. | | logger | Pick<Console, 'debug'> | console | Injected logger object. Must support debug method. | | maxRetries | number | 5 | The number of times to retry a failed batch before throwing an exception. | | throttle | number | 10 | The number of batches to process in parallel. | | unprocessedItemExtractor | (output: Output) => Item[] ⏐ undefined | undefined | Function to extract unprocessed items from an individual batch output. If undefined no retries will be attempted. |

Built for you with ❤️ on Bali! Find more great tools & templates on my GitHub Profile.