Batch Processing
The batch processing utility handles partial failures when processing batches from Amazon SQS, Amazon Kinesis Data Streams, and Amazon DynamoDB Streams.
stateDiagram-v2 direction LR BatchSource: Amazon SQS <br/><br/> Amazon Kinesis Data Streams <br/><br/> Amazon DynamoDB Streams <br/><br/> LambdaInit: Lambda invocation BatchProcessor: Batch Processor RecordHandler: Record Handler function YourLogic: Your logic to process each batch item LambdaResponse: Lambda response BatchSource --> LambdaInit LambdaInit --> BatchProcessor BatchProcessor --> RecordHandler state BatchProcessor { [*] --> RecordHandler: Your function RecordHandler --> YourLogic } RecordHandler --> BatchProcessor: Collect results BatchProcessor --> LambdaResponse: Report items that failed processingKey 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
Background¶
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, orc) when records expire.
journey section Conditions Successful response: 5: Success Maximum retries: 3: Failure Records expired: 1: FailureThis behavior changes when you enable theReportBatchItemFailures feature in your Lambda function event source configuration:
- SQS queues. Only messages reported as failure will return to the queue for a retry, while successful ones will be deleted.
- Kinesis data streams andDynamoDB streams. Single reported failure will use its sequence number as the stream checkpoint. Multiple reported failures will use the lowest sequence number as the checkpoint.
Warning: This utility lowers the chance of processing records more than once; it does not guarantee it
We recommend implementing processing logic in anidempotent manner whenever possible.
You can find more details on how Lambda works with eitherSQS,Kinesis, orDynamoDB in the AWS Documentation.
Getting started¶
Installation¶
Install the library in your project
1 | |
For this feature to work, you need to(1) configure your Lambda function event source to useReportBatchItemFailures, so that the response from the Batch Processing utility can inform the service which records failed to be processed.
Use your preferred deployment framework to set the correct configuration while this utility handles the correct response to be returned.
Required resources¶
The remaining sections of the documentation will rely on these samples. For completeness, this demonstrates IAM permissions and Dead Letter Queue where batch records will be sent after 2 retries.
You do not need any additional IAM permissions to use this utility, except for what each event source requires.
| template.yaml | |
|---|---|
1 2 3 4 5 6 7 8 910111213141516171819202122232425262728293031323334353637383940414243 | |
| template.yaml | |
|---|---|
1 2 3 4 5 6 7 8 91011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556 | |
| template.yaml | |
|---|---|
1 2 3 4 5 6 7 8 9101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566 | |
Processing messages from SQS¶
Processing batches from SQS works in three stages:
- Instantiate
BatchProcessorand chooseEventType.SQSfor the event type - Define your function to handle each batch record, and use the
SQSRecordtype annotation for autocompletion - Use
processPartialResponseto kick off processing
Note
By default, the batch processor will process messages in parallel, which does not guarantee the order of processing. If you need to process messages in order, set theprocessInParallel option tofalse, or useSqsFifoPartialProcessor for SQS FIFO queues.
1 2 3 4 5 6 7 8 91011121314151617181920212223 | |
- Step 1. Creates a partial failure batch processor for SQS queues. Seepartial failure mechanics for details
- Step 2. Defines a function to receive one record at a time from the batch
- Step 3. Kicks off processing
The second record failed to be processed, therefore the processor added its message ID in the response.
1234567 | |
1 2 3 4 5 6 7 8 9101112131415161718192021222324252627282930313233343536 | |
FIFO queues¶
When usingSQS FIFO queues, a batch may include messages from different group IDs.
By default, we will stop processing at the first failure and mark unprocessed messages as failed to preserve ordering. However, this behavior may not be optimal for customers who wish to proceed with processing messages from a different group ID.
Enable theskipGroupOnError option for seamless processing of messages from various group IDs. This setup ensures that messages from a failed group ID are sent back to SQS, enabling uninterrupted processing of messages from the subsequent group ID.
1 2 3 4 5 6 7 8 910111213141516171819202122 | |
1 2 3 4 5 6 7 8 91011121314151617181920212223242526272829303132 | |
Processing messages from Kinesis¶
Processing batches from Kinesis works in three stages:
- Instantiate
BatchProcessorand chooseEventType.KinesisDataStreamsfor the event type - Define your function to handle each batch record, and use the
KinesisStreamRecordtype annotation for autocompletion - Use
processPartialResponseto kick off processing
1 2 3 4 5 6 7 8 9101112131415161718192021 | |
- Creates a partial failure batch processor for Kinesis Data Streams. Seepartial failure mechanics for details
The second record failed to be processed, therefore the processor added its sequence number in the response.
1 2 3 4 5 6 7 8 9101112131415161718192021222324252627282930313233343536 | |
1234567 | |
Processing messages from DynamoDB¶
Processing batches from DynamoDB Streams works in three stages:
- Instantiate
BatchProcessorand chooseEventType.DynamoDBStreamsfor the event type - Define your function to handle each batch record, and use the
DynamoDBRecordtype annotation for autocompletion - Use
processPartialResponseto kick off processing
Info
This code example optionally uses Logger for completion.
1 2 3 4 5 6 7 8 91011121314151617181920212223242526 | |
- Creates a partial failure batch processor for DynamoDB Streams. Seepartial failure mechanics for details
The second record failed to be processed, therefore the processor added its sequence number in the response.
1234567 | |
1 2 3 4 5 6 7 8 91011121314151617181920212223242526272829303132333435363738394041424344454647484950 | |
Error handling¶
By default, we catch any exception raised by your record handler function. This allows us to(1) continue processing the batch,(2) collect each batch item that failed processing, and(3) return the appropriate response correctly without failing your Lambda function execution.
1 2 3 4 5 6 7 8 91011121314151617181920212223242526272829303132 | |
Any exception works here. Seeextending
BatchProcessorsection, if you want to override this behavior.Errors raised in
recordHandlerwill propagate toprocessPartialResponse.
We catch them and include each failed batch item identifier in the response object (seeSample responsetab).
1234567 | |
Partial failure mechanics¶
All records in the batch will be passed to this handler for processing, even if exceptions are thrown - Here's the behaviour after completing the batch:
- All records successfully processed. We will return an empty list of item failures
{'batchItemFailures': []} - Partial success with some exceptions. We will return a list of all item IDs/sequence numbers that failed processing
- All records failed to be processed. We will throw a
FullBatchFailureErrorerror with a list of all the errors thrown while processing unlessthrowOnFullBatchFailureis disabled.
The following sequence diagrams explain how each Batch processor behaves under different scenarios.
SQS Standard¶
Read more aboutBatch Failure Reporting feature in AWS Lambda.
Sequence diagram to explain howBatchProcessor works with SQS Standard queues.
sequenceDiagram autonumber participant SQS queue participant Lambda service participant Lambda function Lambda service->>SQS queue: Poll Lambda service->>Lambda function: Invoke (batch event) Lambda function->>Lambda service: Report some failed messages activate SQS queue Lambda service->>SQS queue: Delete successful messages SQS queue-->>SQS queue: Failed messages return Note over SQS queue,Lambda service: Process repeat deactivate SQS queueSQS FIFO¶
Read more aboutBatch Failure Reporting feature in AWS Lambda.
Sequence diagram to explain howSqsFifoPartialProcessor works with SQS FIFO queues withoutskipGroupOnError flag.
sequenceDiagram autonumber participant SQS queue participant Lambda service participant Lambda function Lambda service->>SQS queue: Poll Lambda service->>Lambda function: Invoke (batch event) activate Lambda function Lambda function-->Lambda function: Process 2 out of 10 batch items Lambda function--xLambda function: Fail on 3rd batch item Lambda function->>Lambda service: Report 3rd batch item and unprocessed messages as failure deactivate Lambda function activate SQS queue Lambda service->>SQS queue: Delete successful messages (1-2) SQS queue-->>SQS queue: Failed messages return (3-10) deactivate SQS queueSequence diagram to explain howSqsFifoPartialProcessor works with SQS FIFO queues withskipGroupOnError flag.
sequenceDiagram autonumber participant SQS queue participant Lambda service participant Lambda function Lambda service->>SQS queue: Poll Lambda service->>Lambda function: Invoke (batch event) activate Lambda function Lambda function-->Lambda function: Process 2 out of 10 batch items Lambda function--xLambda function: Fail on 3rd batch item Lambda function-->Lambda function: Process messages from another MessageGroupID Lambda function->>Lambda service: Report 3rd batch item and all messages within the same MessageGroupID as failure deactivate Lambda function activate SQS queue Lambda service->>SQS queue: Delete successful messages processed SQS queue-->>SQS queue: Failed messages return deactivate SQS queueKinesis and DynamoDB Streams¶
Read more aboutBatch Failure Reporting feature.
Sequence diagram to explain howBatchProcessor works with bothKinesis Data Streams andDynamoDB Streams.
For brevity, we will useStreams to refer to either services. For theory on stream checkpoints, see thisblog post
sequenceDiagram autonumber participant Streams participant Lambda service participant Lambda function Lambda service->>Streams: Poll latest records Lambda service->>Lambda function: Invoke (batch event) activate Lambda function Lambda function-->Lambda function: Process 2 out of 10 batch items Lambda function--xLambda function: Fail on 3rd batch item Lambda function-->Lambda function: Continue processing batch items (4-10) Lambda function->>Lambda service: Report batch item as failure (3) deactivate Lambda function activate Streams Lambda service->>Streams: Checkpoints to sequence number from 3rd batch item Lambda service->>Streams: Poll records starting from updated checkpoint deactivate StreamsThe behavior changes slightly when there are multiple item failures. Stream checkpoint is updated to the lowest sequence number reported.
Note that the batch item sequence number could be different from batch item number in the illustration.
sequenceDiagram autonumber participant Streams participant Lambda service participant Lambda function Lambda service->>Streams: Poll latest records Lambda service->>Lambda function: Invoke (batch event) activate Lambda function Lambda function-->Lambda function: Process 2 out of 10 batch items Lambda function--xLambda function: Fail on 3-5 batch items Lambda function-->Lambda function: Continue processing batch items (6-10) Lambda function->>Lambda service: Report batch items as failure (3-5) deactivate Lambda function activate Streams Lambda service->>Streams: Checkpoints to lowest sequence number Lambda service->>Streams: Poll records starting from updated checkpoint deactivate StreamsAdvanced¶
Parser integration¶
The Batch Processing utility integrates with theParser utility to automatically validate and parse each batch record before processing. This ensures your record handler receives properly typed and validated data, eliminating the need for manual parsing and validation.
To enable parser integration, import theparser function from@aws-lambda-powertools/batch/parser and pass it along with a schema when instantiating theBatchProcessor. This requires you to alsoinstall the Parser utility.
1 | |
You have two approaches for schema validation:
- Item schema only (
innerSchema) - Focus on your payload schema, we handle extending the base event structure - Full event schema (
schema) - Validate the entire event record structure with complete control
Benefits of parser integration¶
Parser integration eliminates runtime errors from malformed data and provides compile-time type safety, making your code more reliable and easier to maintain. Invalid records are automatically marked as failed and won't reach your handler, reducing defensive coding.
12345 | |
123456 | |
Using item schema only¶
When you want to focus on validating your payload without dealing with the full event structure, useinnerSchema. We automatically extend the base event schema for you, reducing boilerplate code while still validating the entire record.
Available transformers by event type:
| Event Type | Base Schema | Available Transformers | When to use transformer |
|---|---|---|---|
| SQS | SqsRecordSchema | json,base64 | json for stringified JSON,base64 for encoded data |
| Kinesis | KinesisDataStreamRecord | base64 | Required for Kinesis data (always base64 encoded) |
| DynamoDB | DynamoDBStreamRecord | unmarshall | Required to convert DynamoDB attribute values to plain objects |
1 2 3 4 5 6 7 8 91011121314151617181920212223242526272829303132333435 | |
1 2 3 4 5 6 7 8 9101112131415161718192021222324252627282930313233343536 | |
Note
IfinnerSchema is used with DynamoDB streams, the schema will be applied to both theNewImage and theOldImage by default. If you want to have dedicated schemas, see the section below.
Using full event schema¶
For complete control over validation, extend the built-in schemas with your custom payload schema. This approach gives you full control over the entire event structure.
1 2 3 4 5 6 7 8 91011121314151617181920212223242526272829303132333435363738 | |
1 2 3 4 5 6 7 8 9101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354 | |
1 2 3 4 5 6 7 8 91011121314151617181920212223242526272829303132333435363738394041424344 | |
Typed record handlers with ParsedRecord¶
To get full type safety in your record handlers, use theParsedRecord utility type:
1234567 | |
This eliminates verbose type annotations and provides clean autocompletion for your parsed data.
Accessing processed messages¶
Use theBatchProcessor directly in your function to access a list of all returned values from yourrecordHandler function.
- When successful. We will include a tuple with
success, the result ofrecordHandler, and the batch record - When failed. We will include a tuple with
fail, exception as a string, and the batch record
| Accessing processed messages | |
|---|---|
1 2 3 4 5 6 7 8 91011121314151617181920212223242526272829 | |
- The processor requires the records array. This is typically handled by
processPartialResponse. - You need to register the
batch, therecordHandlerfunction, and optionally thecontextto access the Lambda context.
Accessing Lambda Context¶
Within yourrecordHandler function, you might need access to the Lambda context to determine how much time you have left before your function times out.
We can automatically inject theLambda context into yourrecordHandler as optional second argument if you pass it to theprocessPartialResponse function.
1 2 3 4 5 6 7 8 910111213141516171819202122232425262728 | |
Working with full batch failures¶
By default, theBatchProcessor will throw aFullBatchFailureError if all records in the batch fail to process, we do this to reflect the failure in your operational metrics.
When working with functions that handle batches with a small number of records, or when you use errors as a flow control mechanism, this behavior might not be desirable as your function might generate an unnaturally high number of errors. When this happens, theLambda service will scale down the concurrency of your function, potentially impacting performance.
For these scenarios, you can set thethrowOnFullBatchFailure option tofalse when calling.
1 2 3 4 5 6 7 8 9101112131415161718 | |
Extending BatchProcessor¶
You might want to bring custom logic to the existingBatchProcessor to slightly override how we handle successes and failures.
For these scenarios, you can subclassBatchProcessor and quickly overridesuccessHandler andfailureHandler methods:
successHandler()– Keeps track of successful batch recordsfailureHandler()– Keeps track of failed batch records
Let's suppose you'd like to add a metric namedBatchRecordFailures for each batch record that failed processing
| Extending failure handling mechanism in BatchProcessor | |
|---|---|
1 2 3 4 5 6 7 8 910111213141516171819202122232425262728293031323334353637383940414243444546 | |
Sequential processing¶
By default, theBatchProcessor processes records in parallel usingPromise.all(). However, if you need to preserve the order of records, you can set theprocessInParallel option tofalse to process records sequentially.
If theprocessInParallel option is not provided, theBatchProcessor will process records in parallel.
When processing records from SQS FIFO queues, we recommend using theSqsFifoPartialProcessor class, which guarantees ordering of records and implements a short-circuit mechanism to skip processing records from a different message group ID.
| Sequential processing | |
|---|---|
1 2 3 4 5 6 7 8 9101112131415161718 | |
Create your own partial processor¶
You can create your own partial batch processor from scratch by inheriting theBasePartialProcessor class, and implementing theprepare(),clean(),processRecord() andprocessRecordSync() abstract methods.
classDiagram direction LR class BasePartialProcessor { <<interface>> +prepare() +clean() +processRecord(record: BaseRecord) +processRecordSync(record: BaseRecord) } class YourCustomProcessor { +prepare() +clean() +processRecord(record: BaseRecord) +processRecordSyc(record: BaseRecord) } BasePartialProcessor <|-- YourCustomProcessor : extendsprepare()– called once as part of the processor initializationclean()– teardown logic called once afterprocessRecordcompletesprocessRecord()– If you need to implement asynchronous logic, use this method, otherwise define it in your class with empty logicprocessRecordSync()– handles all processing logic for each individual message of a batch, including calling therecordHandler(this.handler)
You can then pass this class toprocessPartialResponse to process the records in your Lambda handler function.
| Creating a custom batch processor | |
|---|---|
1 2 3 4 5 6 7 8 910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394 | |
Tracing with AWS X-Ray¶
You can use Tracer to create subsegments for each batch record processed. To do so, you can open a new subsegment for each record, and close it when you're done processing it. When adding annotations and metadata to the subsegment, you can do so directly without callingtracer.setSegment(subsegment). This allows you to work with the subsegment directly and avoid having to either pass the parent subsegment around or have to restore the parent subsegment at the end of the record processing.
1 2 3 4 5 6 7 8 910111213141516171819202122232425262728293031323334353637 | |
- Retrieve the current segment, then create a subsegment for the record being processed
- You can add annotations and metadata to the subsegment directly without calling
tracer.setSegment(subsegment) - Close the subsegment when you're done processing the record
Testing your code¶
As there is no external calls, you can unit test your code withBatchProcessor quite easily.
Example:
Given a SQS batch where the first batch record succeeds and the second fails processing, we should have a single item reported in the function response.
1 2 3 4 5 6 7 8 910111213141516171819202122232425262728293031323334353637383940414243444546 | |
1 2 3 4 5 6 7 8 9101112131415161718192021222324252627 | |
| events/sqs_event.json | |
|---|---|
1 2 3 4 5 6 7 8 9101112131415161718192021222324252627282930313233343536 | |