A simple node script for piping live Amazon Kinesis events to a flat file for analysis and replication

record-kinesis-events.js

var kinesis = require('kinesis');
Transform = require('stream').Transform;
dotenv = require('dotenv');
es = require('event-stream');
fs = require('fs');

dotenv.load();

// Data is retrieved as Record objects, so we transform into Buffers.
var bufferify = new Transform({objectMode: true})
bufferify._transform = function(record, encoding, cb) {
  cb(null, record.Data)
}
// Capture the incoming data and insert newline characters
readStream = kinesis.stream({name: 'video-events'})
  .pipe(bufferify)
  .pipe(es.map(function (data, cb) {
    cb(null, new Buffer(data, 'base64').toString('utf-8') + "\r\n")
  }));

// Send to stdout and a flatfile
readStream.pipe(process.stdout);
readStream.pipe(fs.createWriteStream('test/events.log'));