Кинезис на S3 без пожарного шланга с использованием nodejs lambda

Question

Мы не можем использовать firehose, потому что сообщение в потоке kinesis имеет поле creationDateTime.Исходя из этого, мы хотим, чтобы данные были сброшены в S3.Firehose выдает сообщение на S3 в зависимости от времени прибытия.Итак, у нас есть пользовательская лямбда, которая будет читать 10 000 записей из потока kinesis и помещать их в S3.Код работает нормально, но мы хотим, чтобы сообщения записывались в виде файла .gz.

Вот лямбда-код

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.`;
};

Answer 1

Если вы хотите написать файл .gz, тогда вам придется сжать его самостоятельно.

К счастью, это довольно тривиально в узле:

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) => {
// ...

nb есть асинхронныеверсии методов zlib, если вы хотите сделать это параллельно: https://nodejs.org/docs/latest-v6.x/api/zlib.html#zlib_zlib