How to handle UnprocessedItems using AWS JavaScript SDK (dynamoDB)?

Question

I am trying to use an AWS Lambda function to handle events from SendGrid. As I understand it, the event will be an array with a variable number of JSON objects, each representing a given event. I want to write these events to DynamoDB using batchWriteItem and loop the process until I'm not returned any UnprocessedItems. However, I'm getting stuck in an infinite loop. Here is my code now:

console.log('Loading function');

var aws = require('aws-sdk');
var dynamo = new aws.DynamoDB();
params = {};

exports.handler = function(sg_event, context) {

    var items = [];
    for(var i = 0; i < sg_event.length; i++) {
        var obj = sg_event[i];
        var request = {
            PutRequest: {
                Item: {
                    email: { S: obj.email },
                    timestamp: { S: obj.timestamp.toString() },
                    sg_message_id: { S: obj.sg_message_id },
                    event: { S: obj.event }
                }
            }
        };
        items.push(request);
    }

    params = {
        RequestItems: {
            sendgrid_response: items
        }
    }

    do {
        dynamo.batchWriteItem( params, function(err, data) {
            if(err)
                context.fail(err);
            else
                params.RequestItems = data.UnprocessedItems;
        });
    } while(!isEmpty(params.RequestItems));
};

function isEmpty(obj) {
    return (Object.keys(obj).length === 0);
}

I think the problem is trying to set the params in the callback function, but I don't know how else I'm supposed to do it...I know I could just call another batchWriteItem using UnprocessedItems within the callback of the original one, but I still need to be able to run the function as many times as needed to ensure all UnprocessedItems are written. How can I loop the batchWriteItem correctly?

Dipan · Accepted Answer · 2017-08-29 14:29:27Z

10

@Daniela Miao, Thanks for sharing the solution.

We can add one code block in your posted code which will avoid exception from DynamoDB. This will check if params.RequestItems has Unprocessed data before requesting the DynamoDB for batch write again.

//db is AWS.DynamoDB Client
var processItemsCallback = function(err, data) {
  if (err) { 
     //fail
  } else {
    var params = {};
    params.RequestItems = data.UnprocessedItems;
    /*
    * Added Code block 
    */
    if(Object.keys(params.RequestItems).length != 0) {
      db.batchWriteItem(params, processItemsCallback);
    }
  }
};

db.batchWriteItem(/*initial params*/, processItemsCallback);

answered Aug 29, 2017 at 14:29

Dipan

3112 silver badges7 bronze badges

Sign up to request clarification or add additional context in comments.

Comments

Bing Ren · Accepted Answer · 2018-09-24 03:19:56Z

6

This is my code sample using the "await" syntax. So this code must be in an async function. It does a random delay before retry.

do {
    batchWriteResp = await dynamo.batchWriteItem({RequestItems:batchWriteItems}).promise()
    if (Object.keys(batchWriteResp.UnprocessedItems).length>0) {
        batchWriteItems = batchWriteResp.UnprocessedItems
        // delay a random time between 0.5~2.5 seconds
        const delay = Math.floor(Math.random() * 2000 + 500)
        await new Promise(resolve => setTimeout(resolve, delay));
    } else {
        break
    }
} while (true)

answered Sep 24, 2018 at 3:19

Bing Ren

1,83521 silver badges29 bronze badges

1 Comment

fullstacklife Over a year ago

I found the condition can bork if you don't check for Unprocessed items like so: (batchWriteResp && batchWriteResp.UnprocessedItems && Object.keys(batchWriteResp.UnprocessedItems) && Object.keys(batchWriteResp.UnprocessedItems).length > 0)

Daniela Miao · Accepted Answer · 2015-08-27 01:06:51Z

5

Nodejs is single-threaded and it executes all the main functions first so your while loop will never finish and the callback will never execute.

Here is how you do it:

//db is AWS.DynamoDB Client
var processItemsCallback = function(err, data) {
  if (err) { 
     //fail
  } else {
    var params = {};
    params.RequestItems = data.UnprocessedItems;
    db.batchWriteItem(params, processItemsCallback);
  }
};

db.batchWriteItem(/*initial params*/, processItemsCallback);

answered Aug 27, 2015 at 1:06

Daniela Miao

1,3838 silver badges7 bronze badges

4 Comments

Pimin Konstantin Kefaloukos Over a year ago

Shoud be data.UnprocessedKeys

Kyeotic Over a year ago

@PiminKonstantinKefaloukos No, UnprocessedItems is correct. See the docs

Arek Jaworski Over a year ago

Is there a chance that this can get stuck in infinite loop?

backdesk Over a year ago

You're probably better off using an exponential backoff on db.batchWriteItem if there are unprocessed items.

Gershom Maes · Accepted Answer · 2025-01-16 16:57:10Z

1

I find the accepted answer unnecessarily recursive, and as a result it's very difficult to read and maintain. How about the following?

const isEmptyObj = obj => {
    for (const k of obj) return false;
    return true;
};

async writeDynamodbBatch(db, itemsToWrite) => {
    
    // Note that `itemsToWrite` should look like:
    // { [tableName]: [ item1, item2, item3, ... ] }    
    
    while (!isEmptyObj(itemsToWrite)) {
        
        const result = await db.batchWriteItem({ RequestItems: itemsToWrite }).promise();
        itemsToWrite = result.UnprocessedItems;
        
    }
    
};

edited Jan 16 at 16:57

answered Nov 9, 2023 at 22:03

Gershom Maes

8,2273 gold badges42 silver badges64 bronze badges

Comments

Malone · Accepted Answer · 2019-02-16 14:23:02Z

-1

A pretty neat solution using recursion for BatchGetItem (which works the exact same way as BatchWrite):

    public async batchGet(params: BatchGetItemInput, output: BatchGetResponseMap): Promise<BatchGetResponseMap> {
    const batchGetItemOutput: BatchGetItemOutput = await this.documentClient.batchGet(params).promise();

    Object.keys(batchGetItemOutput.Responses).forEach(tableName => {
        if (output[tableName]) {
            output[tableName] = output[tableName].concat(batchGetItemOutput.Responses[tableName]);
        } else {
            output[tableName] = batchGetItemOutput.Responses[tableName];
        }
    });

    if (Object.keys(batchGetItemOutput.UnprocessedKeys).length !== 0) {
        output = await this.batchGet({ RequestItems: batchGetItemOutput.UnprocessedKeys }, output);
    }

    return output;
}

answered Feb 16, 2019 at 14:23

Malone

1557 bronze badges

1 Comment

Jens Bodal Over a year ago

Errors are also placed into UnprocessedItems. If you do this without a limit on number of retries and you are getting throttled or some other consistent error for an item, you will just exasperate the problem and the loop will go infinitely.

joy · Accepted Answer · 2019-02-28 22:34:22Z

-5

According to AWS documentation, the SDK automatically handles the retry logic:

Note The AWS SDKs implement automatic retry logic and exponential backoff.

And the max number of retries is configurable through:

var dynamodb = new AWS.DynamoDB({apiVersion: '2012-08-10', maxRetries:5});

edited Feb 28, 2019 at 22:34

answered Feb 28, 2019 at 22:25

joy

411 silver badge2 bronze badges

2 Comments

Josh Lee Over a year ago

It does not automatically retry the unprocessed items.

Tarnished-Coder Over a year ago

Also AWS documentation specifies that If batch calls fail partially, failed items will not be retried

Collectives™ on Stack Overflow

How to handle UnprocessedItems using AWS JavaScript SDK (dynamoDB)?

6 Answers 6

Comments

1 Comment

4 Comments

Comments

1 Comment

2 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

6 Answers 6

Comments

1 Comment

4 Comments

Comments

1 Comment

2 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related