Initial commit

Author: abhayachauhan

README.md

@@ -1,2 +1,24 @@
-# change-table-schemas
+# DynamoDB: Change Table Schemas
 A tool to help migrate tables in DynamoDB
+
+## Steps to follow
+
+
+1. Create a new table (let us call this NewTable), with the desired key structure, LSIs, GSIs.
+2. Enable DynamoDB Streams on the original table
+3. Associate a Lambda to the Stream, which pushes the record into NewTable. (This Lambda should trim off the migration flag in Step 5)
+4. [*Optional*] Create a GSI on the original table to speed up scanning items. Ensure this GSI only has attributes: Primary Key, and Migrated (See Step 5).
+5. Scan the GSI created in the previous step (or entire table) and use the following Filter:
+
+    FilterExpression = "attribute_not_exists(Migrated)"
+
+    Update each item in the table with a migrate flag (ie: “Migrated”: { “S”: “0” }, which sends it to the DynamoDB Streams (using UpdateItem API, to ensure no data loss occurs).
+
+    **NOTE** You may want to increase write capacity units on the table during the updates.
+6. The Lambda will pick up all items, trim off the Migrated flag and push it into NewTable.
+7. Once all items have been migrated, repoint the code to the new table
+8. Remove original table, and Lambda function once happy all is good.
+
+## More information
+
+https://www.abhayachauhan.com/2017/10/dynamodb-changing-table-schema/

flag-for-migration/flagForMigration.js

@@ -0,0 +1,51 @@
+var async = require('async');
+
+exports.flagForMigration = (dynamodb, tableName, migrationAttributeName, itemKeys) => {
+    return new Promise((resolve, reject) => {
+        function shuffle(a) {
+            var j, x, i;
+            for (i = a.length - 1; i > 0; i--) {
+                j = Math.floor(Math.random() * (i + 1));
+                x = a[i];
+                a[i] = a[j];
+                a[j] = x;
+            }
+        }
+
+        function updateItem(item, cb) {
+            delete item['migrationFlag'];
+            console.log(item);
+            var params = {
+                "Key": item,
+                "TableName": tableName,
+                "ExpressionAttributeNames": {
+                    "#MF": migrationAttributeName
+                },
+                "ExpressionAttributeValues": {
+                    ":t": { BOOL: true }
+                },
+                "UpdateExpression": "SET #MF = :t",
+                "ReturnConsumedCapacity": "INDEXES"
+            }
+
+            dynamodb.updateItem(params, function(err, data) {
+                if (err)
+                    console.log('Error', params);
+                if (!err)
+                    itemKeys.splice(itemKeys.indexOf(item));
+                cb(err, data);
+            });
+        }
+        
+        // Shuffle keys so we minimise hot partition keys
+        shuffle(itemKeys);
+
+        async.eachLimit(itemKeys, 2, updateItem, function(err) {
+            if (err) {
+                reject(err);
+            }
+            resolve();
+        });
+    });
+}
+

flag-for-migration/index.js

@@ -0,0 +1,20 @@
+var AWS = require('aws-sdk');
+var dynamodb = new AWS.DynamoDB();
+var async = require('async');
+var scan = require('./scanTable.js');
+var migrate = require('./flagForMigration.js');
+
+var tableName = "XXX"; // TABLE NAME HERE
+var gsiName = "XXX"; // GSI NAME HERE
+var migrationAttributeName = "migrationFlag"; // MIGRATION ATTRIBUTE HERE
+
+startWork();
+
+function startWork() {
+    scan.scanTable(dynamodb, tableName, gsiName, migrationAttributeName)
+        .then(itemKeys => {
+            console.log('Finished scanning, found', itemKeys.length,'items.');
+            
+            migrate.flagForMigration(dynamodb, tableName, migrationAttributeName, itemKeys);
+        });
+}

flag-for-migration/scanTable.js

@@ -0,0 +1,39 @@
+exports.scanTable = (dynamodb, tableName, gsiIndexName, migrationAttributeName) => {
+    var items = [];
+
+    return new Promise((resolve, reject) => {
+
+        function iterateTable(params) {
+            var promise = dynamodb.scan(params).promise();
+            
+            promise
+                .then(data => {
+                    console.log('Received',data.Items.length,'items');
+                    items = items.concat(data.Items);
+                    if (data.LastEvaluatedKey) {
+                        params.ExclusiveStartKey = data.LastEvaluatedKey;
+                        iterateTable(params);
+                    } else {
+                        resolve(items);
+                    }
+                })
+                .catch(err => {
+                    console.log('ERROR:');
+                    console.log(err);
+                    iterateTable(params);
+                });    
+        }
+
+        var params = {
+            "TableName": tableName,
+            "ReturnConsumedCapacity": "INDEXES"
+        }
+        if (gsiIndexName)
+            params.IndexName = gsiIndexName;
+        if (migrationAttributeName)
+            params.FilterExpression = "attribute_not_exists(" + migrationAttributeName + ")";
+    
+        iterateTable(params);
+
+    });
+}

migration-lambda/index.js

@@ -0,0 +1,36 @@
+var AWS = require('aws-sdk');
+var dynamodb = new AWS.DynamoDB();
+var newTableName = process.env.destination;
+
+exports.handler = (event, context, callback) => {
+    var promises = [];
+
+    event.Records.forEach(function(record) {
+      if (record.eventName === "REMOVE") {
+          var params = {
+            Key: record.dynamodb.OldImage.Key,
+            TableName: newTableName
+          };
+
+          promises.push(
+            dynamodb.deleteItem(params).promise()
+          );
+
+        }
+      else {
+        var params = {
+            Item: record.dynamodb.NewImage,
+            TableName: newTableName
+          };
+
+          promises.push(
+            dynamodb.putItem(params).promise()
+          );
+      }
+      
+    });
+
+    Promise.all(promises)
+      .then(() => callback())
+      .catch((err) => callback(err));
+};

migration-lambda/lambda-iam-policy.json

@@ -0,0 +1,45 @@
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": [
+        "lambda:InvokeFunction"
+      ],
+      "Resource": [
+        "*"
+      ]
+    },
+    {
+      "Effect": "Allow",
+      "Action": [
+        "logs:CreateLogGroup",
+        "logs:CreateLogStream",
+        "logs:PutLogEvents"
+      ],
+      "Resource": "arn:aws:logs:*:*:*"
+    },
+    {
+      "Effect": "Allow",
+      "Action": [
+        "dynamodb:PutItem",
+        "dynamodb:DeleteItem"
+      ],
+      "Resource": [
+          "arn:aws:dynamodb:*:*:table/*"
+      ]
+    },
+    {
+      "Effect": "Allow",
+      "Action": [
+        "dynamodb:DescribeStream",
+        "dynamodb:GetRecords",
+        "dynamodb:GetShardIterator",
+        "dynamodb:ListStreams"
+      ],
+      "Resource": [
+        "arn:aws:dynamodb:*:*:table/*/stream/*"
+      ]
+    }
+  ]
+}

package.json

@@ -0,0 +1,23 @@
+{
+  "name": "change-table-schemas",
+  "version": "1.0.0",
+  "description": "A tool to help migrate tables in DynamoDB",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "repository": {
+    "type": "git",
+    "url": "git+https://github.com/abhayachauhan/change-table-schemas.git"
+  },
+  "author": "Abhaya Chauhan",
+  "license": "MIT",
+  "bugs": {
+    "url": "https://github.com/abhayachauhan/change-table-schemas/issues"
+  },
+  "homepage": "https://github.com/abhayachauhan/change-table-schemas#readme",
+  "dependencies": {
+    "async": "^2.6.0",
+    "aws-sdk": "^2.176.0"
+  }
+}