Add script to migrate data from S3 to QLDB

Author: xxcxy

README.md

@@ -77,6 +77,12 @@ Make sure `QLDB_NAME`, `AWS_REGION`, `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY
 
 Run `npm run migrate-qldb-to-pg` to import data from qldb. 
 
+## Import data from S3 to QLDB
+
+Make sure `BUCKET_NAME`, `QLDB_NAME`, `AWS_REGION`, `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY` are set in your environment
+
+Run `npm run import-s3-data` to import data from s3 to qldb. 
+
 ## Local Deployment with Docker
 
 Make sure all config values are right, and you can run on local successfully, then run below commands

package.json

@@ -10,7 +10,8 @@
     "migrate-db-to-es": "node scripts/db/dumpDbToEs.js",
     "migrations": "node scripts/db/migrations.js",
     "insert-data": "node scripts/db/insert-data.js",
-    "migrate-qldb-to-pg": "node scripts/db/migrateQldbToPg.js"
+    "migrate-qldb-to-pg": "node scripts/db/migrateQldbToPg.js",
+    "import-s3-data": "node scripts/db/importS3ToQldb.js"
   },
   "repository": {
     "type": "git",

scripts/db/importS3ToQldb.js

@@ -0,0 +1,90 @@
+const _ = require('lodash')
+const ion = require('ion-js')
+const AWS = require('aws-sdk')
+const QLDB = require('amazon-qldb-driver-nodejs')
+const logger = require('../../src/common/logger')
+
+// Create S3 service object
+const s3 = new AWS.S3({
+  apiVersion: '2006-03-01',
+  region: process.env.AWS_REGION,
+  accessKeyId: process.env.AWS_ACCESS_KEY_ID,
+  secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY
+})
+
+/**
+ * Get the statement table.
+ * @param {String} statement the execute statement
+ */
+function getTable (statement) {
+  if (statement.startsWith('create index on') || statement.startsWith('CREATE INDEX ON')) {
+    return statement.substring(16, statement.indexOf(' ', 16))
+  }
+  if (statement.startsWith('create table')) {
+    return statement.substring(13)
+  }
+}
+
+/**
+ * import data from s3 to qldb.
+ */
+async function main () {
+  const qldbInstance = new QLDB.QldbDriver(process.env.QLDB_NAME, {
+    region: process.env.AWS_REGION,
+    accessKeyId: process.env.AWS_ACCESS_KEY_ID,
+    secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY
+  })
+  const tables = await qldbInstance.getTableNames()
+  const bucket = process.env.BUCKET_NAME
+
+  try {
+    const list = await s3.listObjects({ Bucket: bucket }).promise()
+    // get all data file key
+    const keys = _(list.Contents).map('Key').filter(key => _.endsWith(key, '.ion'))
+      .sortBy(key => _.parseInt(_.split(_.split(key, '.')[1], '-')[0])).value()
+    for (const key of keys) {
+      // get data of each file
+      const data = await s3.getObject({ Bucket: bucket, Key: key }).promise()
+      const contents = ion.loadAll(data.Body)
+      for (const content of contents) {
+        // get content of each transaction
+        const statement = content.transactionInfo.statements.get(0).statement.stringValue()
+        // process create statement
+        if (statement.startsWith('create') || statement.startsWith('CREATE')) {
+          if (!_.includes(tables, getTable(statement))) {
+            logger.info(`start executing: ${statement}`)
+            await qldbInstance.executeLambda(t => t.execute(statement))
+          } else {
+            logger.info(`exists, skip ${statement}`)
+          }
+        }
+        // process insert and update statement
+        if (statement.startsWith('INSERT') || statement.startsWith('UPDATE')) {
+          logger.info(`start executing: ${statement}`)
+          try {
+            await qldbInstance.executeLambda(t => t.execute(statement, content.revisions.get(0).data))
+          } catch (e) {
+            logger.error(e)
+            logger.warn(`execute ${statement} failed`)
+          }
+        }
+        // process delete statement
+        if (statement.startsWith('DELETE')) {
+          logger.info(`start executing: ${statement}`)
+          try {
+            await qldbInstance.executeLambda(t => t.execute(statement))
+          } catch (e) {
+            logger.error(e)
+            logger.warn(`execute ${statement} failed`)
+          }
+        }
+      }
+    }
+  } finally {
+    qldbInstance.close()
+  }
+}
+
+(async () => {
+  main().catch(err => console.error(err))
+})()