Node Kinesis Client Library
Based on the AWS Kinesis Client Library for Java, reimplemented in Node.js.
- Build Kinesis consumers easily
- Automatically scale up or down as streams split or merge
- Allow distributed processing without any extra code
Install with npm install kinesis-client-library --save
.
I am using this in production, but it's very new so there's a good chance it will (1) have some bugs and (2) go through a breaking change at some point. Feedback welcome, use with care.
Why?
I had a stack in Node.js. Working with and deploying a Java app just to consume a stream was too inconvenient, so I made this.
Terminology
- Consumer: A single process that is responsible for consuming a single Kinesis shard. Consumers are created and managed by clusters; they should not be created directly directly.
- Cluster: A process that manages consumers. A single cluster can have many consumers (or 1, or 0) that all run on the same machine. Clusters should be created by the CLI.
- Network: All of the consumers processing a single Kinesis stream for a single application. Networks can have many clusters that are run on one or many machines.
Networks are meant to be distributed (though they don't have to be) and will automatically rebalance shards across clusters. Each cluster will try to process an equal number of shards, so it's a good idea to make sure each cluster has equivalent resources (e.g. clusters should have similar memory and CPU available). Networks will automatically pick up new shards as splits or merges happen.
Multiple networks can independently process the same stream as long as the stream has enough capacity. In this case, you should be sure to give each set of processors a unique table
flag.
Usage
Before you can do anything with this library, you need to create a Kinesis stream. Everything from this point forward assumes you have an existing stream.
CLI
Clusters are designed to be launched from the command line. The library exposes a launch-kinesis-cluster
executable.
$ launch-kinesis-cluster --help
Usage:
--help (Display this message)
Required flags:
--consumer [Path to consumer file]
--table [DynamoDB table name]
--stream [Kinesis stream name]
Optional flags:
--start-at [Starting iterator type] ("trim_horizon" or "latest", defaults to "trim_horizon")
--capacity.[read|write] [Throughput] (DynamoDB throughput for *new* tables, defaults to 10 for each)
--aws.[option] [Option value] (e.g. --aws.region us-west-2)
--http [port] (Start HTTP server, port defaults to $PORT)
--log-level [level] (Logging verbosity, uses Bunyan log levels)
--dynamo-endpoint (Use a cusotm endpoint for the DynamoDB service)
--local-dynamo (Whether or not to use a local implementation of DynamoDB, defaults to false)
--local-dynamo-directory (Directory to store local DB, defaults to temp directory)
--kinesis-endpoint (Use a custom endpoint for the Kinesis service)
--local-kinesis (Use a local implementation of Kinesis, defaults to false)
--local-kinesis-port (Port to access local Kinesis on, defaults to 4567)
--local-kinesis-no-start (Assume a local Kinesis server is already running, defaults to false)
--num-records (Maximum number of records to get in each Kinesis query, defaults to the Kinesis maximum of 10000)
Notes:
table
is used as a DynamoDB table name that is unique to the network. If the table doesn't exist yet, it is created.stream
must match a Kinesis stream that already exists.- The (optional) HTTP server is primarily meant to be used as a health check.
- AWS options are based to the
AWS.Service
constructor, so any other credential strategies (e.g. environment variables, EC2 IAM roles) will be used automatically.
Consumer
Consumers are implemented in JavaScript by calling AbstractConsumer.extend()
with an object that implements some/all of these methods:
processRecords
: Accepts an array of Record objects (Record.Data
will be a Buffer
object) and a callback. Pass true
as the second argument to the callback when you want to save a checkpoint — i.e. when you have processed a chunk of data.processResponse
: Accepts a full response from Kinesis' GetRecords API and a callback. Pass true
as the second argument to the callback when you want to save a checkpoint — i.e. when you have processed a chunk of data.initialize
: Called when a consumer is spawned, before any records are processed. Accepts a callback that must be called to start record processing.shutdown
: Called when a consumer is about to exit. Accepts a callback that must be called to complete shutdown; if the callback is not called without 30 seconds the process exits anyway.
A consumer MUST implement either processRecords
or processResponse
.
Example consumer implementation
This consumer uploads records to S3 in 50 megabyte batches.
var s3 = new require('aws-sdk').S3()
var kcl = require('kinesis-client-library')
var newlineBuffer = new Buffer('\n')
kcl.AbstractConsumer.extend({
initialize: function (done) {
this.cachedRecords = []
this.cachedRecordsSize = 0
done()
},
processRecords: function (records, done) {
records.forEach(function (record) {
this.cachedRecords.push(record.Data)
this.cachedRecords.push(newlineBuffer)
this.cachedRecordsSize += (record.Data.length + newlineBuffer.length)
}.bind(this))
var shouldCheckpoint = this.cachedRecordsSize > 50000000
if (! shouldCheckpoint) return done()
s3.putObject({
Bucket: 'my-bucket-name',
Key: 'path/to/records/' + Date.now(),
Body: Buffer.concat(this.cachedRecords)
}, function (err) {
if (err) return done(err)
this.cachedRecords = []
this.cachedRecordsSize = 0
done(null, true)
}.bind(this))
}
})
Example consumer launch
Assume the code above is in lib/consumer.js
and you've got a package with kinesis-client-library
declared as a dependency.
For convenience, we'll assume your AWS credentials are defined as environment variables. Using the command line, you could launch a cluster like this.
launch-kinesis-cluster \
--consumer lib/consumer.js \
--table MyKinesisConsumerApp \
--stream MyKinesisStreamName \
--aws.region us-east-1