You've successfully subscribed to MyPad Blog
Great! Next, complete checkout for full access to MyPad Blog
Welcome back! You've successfully signed in.
Success! Your account is fully activated, you now have access to all content.
Success! Your billing info is updated.
Billing info update failed.

Daily Coding - Write data from an AWS Lambda call to sqlite3

Daily Coding - Write data from an AWS Lambda call to sqlite3

In this post I will explain the process of writing data from an AWS lambda call to a local sqlite database.

The use case I have is as follows. I have an AWS lambda function that runs a scraping job to get data from a service. I'd like the lambda call to be triggered from my server machine using a cron job at certain points.

Once I get the response from the lambda function, I would like to use the data returned from the AWS Lambda call and then parse and write the data to a local sequel database using Node.js.

The following script accomplishes the same. I'm using sqlite3 and sequelize from npm.

Further as part of this demonstration, it shows the utility of using a lambda function to run parallel independent jobs. And aggregate the data once the responses are received from the function.

'use strict';
require('dotenv').config();

var AWS = require('aws-sdk');
const fs = require('fs');
const rison = require('rison');

const moment = require("moment");
const _ = require('underscore');
const async = require("async");

const uuidv4 = require('uuid/v4');

var Sequelize = require("sequelize");
var sequelize = new Sequelize('main', '', '', {
    // sqlite! now!
    dialect: 'sqlite',

    // the storage engine for sqlite
    // - default ':memory:'
    storage: './data/'+moment().format('YYYYMMDD')+'-marketwatch'+'.sqlite3',
    logging:false
});

const Option = sequelize.define('option', {
    call_ask: Sequelize.NUMBER,
    call_bid: Sequelize.NUMBER,
    call_change: Sequelize.NUMBER,
    call_last: Sequelize.NUMBER,
    call_openint: Sequelize.NUMBER,
    call_quote: Sequelize.STRING,
    call_vol: Sequelize.NUMBER,
    expiry: Sequelize.STRING,
    put_ask: Sequelize.NUMBER,
    put_bid: Sequelize.NUMBER,
    put_change: Sequelize.NUMBER,
    put_last: Sequelize.NUMBER,
    put_openint: Sequelize.NUMBER,
    put_quote: Sequelize.STRING,
    put_vol: Sequelize.NUMBER,
    strike: Sequelize.NUMBER,
    ticker: Sequelize.STRING,
    underlying: Sequelize.NUMBER,
    timestamp: Sequelize.STRING,
    timestamp_int: Sequelize.NUMBER,
    id: { type: Sequelize.STRING, primaryKey: true }
}, {

});

sequelize
    .authenticate()
    .then(function(err) {
        console.log('Connection has been established successfully.');
    }, function(err) {
        console.log('Unable to connect to the database:', err);
    });


var timestamp = moment().toISOString();
var timestamp_int = moment().unix();

async function writeResult(result) {
    let data = result.Payload;

    // remove the quotes
    data = String(data);
    data = data.substring(1, data.length - 1);
    var options = rison.decode_array(data);

    // console.log(options);

    var items = [];

    _.each(options, async function(o) {
        o.underlying = Number(o.underlying);
        o.expiry = moment(o.expiry).toDate().toISOString();
        o.timestamp = timestamp;
        o.timestamp_int = timestamp_int;
        o.id = uuidv4();

        // console.log(o);
        items.push(o);

        await Option.create(o);
    });

    // console.log(items);
}

const invokeLambda = (lambda, params) => new Promise((resolve, reject) => {
    lambda.invoke(params, (error, data) => {
        if (error) {
            reject(error);
        } else {
            resolve(data);
        }
    });
});

(async () => {

    let region = process.env.AWS_REGION;
    let functionURN = process.env.AWS_FUNCTION_URN;

    // You shouldn't hard-code your keys in production!
    // http://docs.aws.amazon.com/AWSJavaScriptSDK/guide/node-configuring.html
    AWS.config.update({
        accessKeyId: process.env.AWS_ACCESS_KEY,
        secretAccessKey: process.env.AWS_SECRET_KEY,
        region: region,
    });

    const lambda = new AWS.Lambda();

    let keywords = ['ASPS', 'FB', 'TSLA', 'OCN', 'PTON', 'TWTR', 'SPOT'];
    // keywords = ['ASPS'];
    // keywords = ['ASPS', 'FB'];

    let promises = [];

    for (let kw of keywords) {
        let event = { keyword: kw };
        let params = {
            FunctionName: functionURN,
            InvocationType: "RequestResponse",
            Payload: JSON.stringify(event),
        };
        console.log(params);
        promises.push(
            invokeLambda(lambda, params)
        )
    }

    console.log(`Invoked ${promises.length} lambda requests!`);

    var start = new Date();
    let results = await Promise.all(promises);
    var end = new Date() - start;

    console.log(`invokeLambda() in region ${region} took ${end/1000} seconds`);

    await sequelize.sync({});
    console.log("All models were synchronized successfully.");

    const initialTimestamp = new Date();

    await Promise.all(
        results.map(async result => {
            await writeResult(result);
        })
    );

    console.log(
        `All Completed! ${Number(new Date()) - Number(initialTimestamp)}ms.`
    );
})();