Using AWS Lambda with Amazon Kinesis with Batch Item Handling.
The Lambda function retrieves Kinesis Records from the event parameter that it receives. The function retrieves Kinesis payload, decode from Base64 and log them to cloudwatch. The function reports batchItemFailures with SequenceNumber, in order for this feature to work ReportBatchItemFailures should be enabled in lambda configurations
Select language:
Usage Example with Node:
Consuming Kinesis event with Lambda using JavaScript with batch item handling.
exports.handler = async (event, context) => {
for (const record of event.Records) {
try {
console.log(`Processed Kinesis Event - EventID: ${record.eventID}`);
const recordData = await getRecordDataAsync(record.kinesis);
console.log(`Record Data: ${recordData}`);
// TODO: Do interesting work based on the new data
} catch (err) {
console.error(`An error occurred ${err}`);
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
return {
batchItemFailures: [{ itemIdentifier: record.kinesis.sequenceNumber }],
};
}
}
console.log(`Successfully processed ${event.Records.length} records.`);
return { batchItemFailures: [] };
};
async function getRecordDataAsync(payload) {
var data = Buffer.from(payload.data, "base64").toString("utf-8");
await Promise.resolve(1); //Placeholder for actual async work
return data;
}
Created by:
