@aws-lambda-powertools/batch
TypeScript icon, indicating that this package has built-in type declarations

2.0.4 • Public • Published

Powertools for AWS Lambda (TypeScript) - Batch Processing Utility

Powertools for AWS Lambda (TypeScript) is a developer toolkit to implement Serverless best practices and increase developer velocity.

You can use the package in both TypeScript and JavaScript code bases.

Intro

The batch processing utility handles partial failures when processing batches from Amazon SQS, Amazon Kinesis Data Streams, and Amazon DynamoDB Streams.

Key features

  • Reports batch item failures to reduce number of retries for a record upon errors
  • Simple interface to process each batch record
  • Build your own batch processor by extending primitives

Usage

To get started, install the library by running:

npm install @aws-lambda-powertools/batch

Batch Processor

When using SQS, Kinesis Data Streams, or DynamoDB Streams as a Lambda event source, your Lambda functions are triggered with a batch of messages.

If your function fails to process any message from the batch, the entire batch returns to your queue or stream. This same batch is then retried until either condition happens first: a) your Lambda function returns a successful response, b) record reaches maximum retry attempts, or c) when records expire.

With this utility, batch records are processed individually – only messages that failed to be processed return to the queue or stream for a further retry.

SQS Processor

When using SQS as a Lambda event source, you can specify the EventType.SQS to process the records. The response will be a SQSBatchResponse which contains a list of items that failed to be processed.

import {
  BatchProcessorSync,
  EventType,
  processPartialResponseSync,
} from '@aws-lambda-powertools/batch';
import { Logger } from '@aws-lambda-powertools/logger';
import type {
  SQSEvent,
  SQSRecord,
  Context,
  SQSBatchResponse,
} from 'aws-lambda';

const processor = new BatchProcessorSync(EventType.SQS);
const logger = new Logger();

const recordHandler = (record: SQSRecord): void => {
  const payload = record.body;
  if (payload) {
    const item = JSON.parse(payload);
    logger.info('Processed item', { item });
  }
};

export const handler = async (
  event: SQSEvent,
  context: Context
): Promise<SQSBatchResponse> => {
  return processPartialResponseSync(event, recordHandler, processor, {
    context,
  });
};
export { processor };

Kinesis Processor

When using Kinesis Data Streams as a Lambda event source, you can specify the EventType.KinesisDataStreams to process the records. The response will be a KinesisStreamBatchResponse which contains a list of items that failed to be processed.

import {
  BatchProcessorSync,
  EventType,
  processPartialResponseSync,
} from '@aws-lambda-powertools/batch';
import { Logger } from '@aws-lambda-powertools/logger';
import type {
  KinesisStreamEvent,
  KinesisStreamRecord,
  Context,
  KinesisStreamBatchResponse,
} from 'aws-lambda';

const processor = new BatchProcessorSync(EventType.KinesisDataStreams);
const logger = new Logger();

const recordHandler = (record: KinesisStreamRecord): void => {
  logger.info('Processing record', { record: record.kinesis.data });
  const payload = JSON.parse(record.kinesis.data);
  logger.info('Processed item', { item: payload });
};

export const handler = async (
  event: KinesisStreamEvent,
  context: Context
): Promise<KinesisStreamBatchResponse> => {
  return processPartialResponseSync(event, recordHandler, processor, {
    context,
  });
};

DynamoDB Streams Processor

When using DynamoDB Streams as a Lambda event source, you can use the BatchProcessorSync with the EventType.DynamoDBStreams to process the records. The response will be a DynamoDBBatchResponse which contains a list of items that failed to be processed.

import {
  BatchProcessorSync,
  EventType,
  processPartialResponseSync,
} from '@aws-lambda-powertools/batch';
import { Logger } from '@aws-lambda-powertools/logger';
import type {
  DynamoDBStreamEvent,
  DynamoDBRecord,
  Context,
  DynamoDBBatchResponse,
} from 'aws-lambda';

const processor = new BatchProcessorSync(EventType.DynamoDBStreams);
const logger = new Logger();

const recordHandler = (record: DynamoDBRecord): void => {
  if (record.dynamodb && record.dynamodb.NewImage) {
    logger.info('Processing record', { record: record.dynamodb.NewImage });
    const message = record.dynamodb.NewImage.Message.S;
    if (message) {
      const payload = JSON.parse(message);
      logger.info('Processed item', { item: payload });
    }
  }
};

export const handler = async (
  event: DynamoDBStreamEvent,
  context: Context
): Promise<DynamoDBBatchResponse> => {
  return processPartialResponseSync(event, recordHandler, processor, {
    context,
  });
};

Async processing

If your use case allows you to process multiple records at the same time without conflicting with each other, you can use the BatchProcessor to process records asynchronously. This will create an array of promises that will be resolved once all records have been processed.

import {
  BatchProcessor,
  EventType,
  processPartialResponse,
} from '@aws-lambda-powertools/batch';
import type {
  SQSEvent,
  SQSRecord,
  Context,
  SQSBatchResponse,
} from 'aws-lambda';

const processor = new BatchProcessor(EventType.SQS);

const recordHandler = async (record: SQSRecord): Promise<number> => {
  const res = await fetch('https://httpbin.org/anything', {
    body: JSON.stringify({ message: record.body }),
  });

  return res.status;
};

export const handler = async (
  event: SQSEvent,
  context: Context
): Promise<SQSBatchResponse> => {
  return await processPartialResponse(event, recordHandler, processor, {
    context,
  });
};

Check the docs for more examples.

Contribute

If you are interested in contributing to this project, please refer to our Contributing Guidelines.

Roadmap

The roadmap of Powertools for AWS Lambda (TypeScript) is driven by customers’ demand.
Help us prioritize upcoming functionalities or utilities by upvoting existing RFCs and feature requests, or creating new ones, in this GitHub repository.

Connect

How to support Powertools for AWS Lambda (TypeScript)?

Becoming a reference customer

Knowing which companies are using this library is important to help prioritize the project internally. If your company is using Powertools for AWS Lambda (TypeScript), you can request to have your name and logo added to the README file by raising a Support Powertools for AWS Lambda (TypeScript) (become a reference) issue.

The following companies, among others, use Powertools:

Sharing your work

Share what you did with Powertools for AWS Lambda (TypeScript) 💞💞. Blog post, workshops, presentation, sample apps and others. Check out what the community has already shared about Powertools for AWS Lambda (TypeScript) here.

Using Lambda Layer

This helps us understand who uses Powertools for AWS Lambda (TypeScript) in a non-intrusive way, and helps us gain future investments for other Powertools for AWS Lambda languages. When using Layers, you can add Powertools as a dev dependency (or as part of your virtual env) to not impact the development process.

Credits

Credits for the Lambda Powertools for AWS Lambda (TypeScript) idea go to DAZN and their DAZN Lambda Powertools.

License

This library is licensed under the MIT-0 License. See the LICENSE file.

Package Sidebar

Install

npm i @aws-lambda-powertools/batch

Weekly Downloads

10,095

Version

2.0.4

License

MIT-0

Unpacked Size

86.5 kB

Total Files

72

Last publish

Collaborators

  • aws-powertools-bot
  • am29d
  • dreamorosi