Kinesis to S3 without firehose using nodejs lambda

Question

We cant use firehose because message in kinesis stream has a "creationDateTime" field. Based on this we want the data to be dumped in S3. Firehose would dump the message on S3 based on the arrival time. So, we have a custom lambda that will read 10,000 records from kinesis stream and put those in S3. Code is working fine but we want the messages written as .gz file.

Here is lambda code

console.log('Loading function');
const AWS = require('aws-sdk');

const awsConfig = {
  region: 'us-west-2',
  apiVersion: '2012-08-10',
};
AWS.config.update(awsConfig);
const s3 = new AWS.S3();
const bucket = 'uis-prime-test';
// const uniqueId = Math.floor(Math.random() * 100000);


// initially create the map without any key
const map = {};
function addValueToList(key, value) {
  // if the list is already created for the "key", then uses it
  // else creates new list for the "key" to store multiple values in it.
  map[key] = map[key] || [];
  map[key].push(value);
}

function getS3Key(payload) {
  const json = JSON.parse(payload);
  const creationDateTime = new Date(json.executionContext.creationDateTime);
  const year = creationDateTime.getUTCFullYear();
  let month = creationDateTime.getUTCMonth() + 1;
  const day = creationDateTime.getUTCDate();
  const hour = creationDateTime.getUTCHours();
  if (month < 10) { month = `0${month}`; }
  return `${year}/${month}/${day}/${hour}/`;
}

exports.handler = function (event, context) {
  try {
    const uniqueId = context.awsRequestId;
    event.Records.forEach((record) => {
      const payload = Buffer.from(record.kinesis.data, 'base64').toString('ascii');
      const key = getS3Key(payload) + uniqueId;
      addValueToList(key, payload.toString());
    });

    Object.entries(map).forEach(([key, value]) => {
      const params = { Bucket: bucket, Key: key, Body: value.join('\n') };
      s3.putObject(params, (err, data) => {
        if (err) {
          throw err;
        } else {
          console.log('Successfully uploaded data');
        }
      });
    });
  } catch (err) {
    console.log(err);
  }
  return `Successfully processed ${event.Records.length} records.`;
};

thomasmichaelwallace thomasmichaelwallace · Accepted Answer · 2018-10-10T08:59:39

If you want to write a .gz file then you'll have to gzip it yourself.

Luckily that's fairly trivial to do in node:

const zlib = require('zlib'); // this is a node build-in so no packages required
// ...
    Object.entries(map).forEach(([key, value]) => {
      const Body = zlib.gzipSync(value.join('\n'));
      const params = { Bucket: bucket, Key: key, Body };
      s3.putObject(params, (err, data) => {
// ...

n.b. there are asyncronous versions of the zlib methods if you want to do this in parallel: https://nodejs.org/docs/latest-v6.x/api/zlib.html#zlib_zlib

Kinesis to S3 without firehose using nodejs lambda

1 Answers