Comprehensive guide to Redis. Part 3.

vladimir

November 8, 2015

Comprehensive guide to Redis. Part 3.

A time series is an ordered sequence of values that are made over a time interval. You can use time series in statistics, communications, and social networks. I this tutorial we are going to create a simple stock time series Node.js library using Redis Strings. This library records events per second, minute, hour, and day.

This library will be able to save an event at a given timestamp with an insert method and fetch values within a range of timestamps with a fetch method. The library we are going to create provides multiple granularities: second, minute, hour, and day. For example, if an event happens on date 8/11/2015 at 00:00:00(timestamp 1446940800), the following Redis keys will be incremented:
– event:1sec:1446940800
– event:1min:1446940800
– event:1hour:1446940800
– event:1day:1446940800

Create a file stock-timeseries.js with the following code:

//time series constructor. Requires a redis client and a namespace
function TimeSeries(client, namespace){

    this.namespace = namespace;
    this.client = client;
    //granularity names and their equivalents in seconds
    this.units = {
        second:1,
        minute: 60,
        hour: 60 * 60,
        day: 24 * 60 * 60
    };
    //each granularity has a name, TTL(time to live) and a duration.
    //the null ttl present on 1dat meands that this ttl never expires
    this.granularities = {
        '1sec': {name: '1sec', ttl: this.units.hour * 2, duration: this.units.second},
        '1min': {name: '1min', ttl: this.units.day * 7, duration: this.units.minute},
        '1hour': {name: '1hour', ttl: this.units.day * 60, duration: this.units.hour},
        '1day': {name: '1day', ttl: null, duration: this.units.day}
    };

}

//insert a particular price at a given timestamp
TimeSeries.prototype.insert = function(timestampInSeconds, price){
    //iterate over all franularities
    for (var granularityName in this.granularities){
        var granularity = this.granularities[granularityName];
        //get a key name in the format "napespace:granularity:timestamp"
        //for ex.: "google:1sec:12"
        var key = this._getKeyName(granularity, timestampInSeconds);
        //execute the SET command
        this.client.set(key, price);
        //the EXPIRE command
        //pass the key and the ttl
        //this command deletes a redis key automatically after a given number
        //of seconds
        if(granularity.ttl !== null){
            this.client.expire(key, granularity.ttl);
        }
    }
};

//returns a key based on granularitu and timestamp
TimeSeries.prototype._getKeyName = function(granularity, timestampInSeconds){
    var roundedTimestamp = this._getRoundedTimestamp(timestampInSeconds, granularity.duration);
    return [this.namespace, granularity.name, roundedTimestamp].join(':');
};

//returns a normalized timestamp by granularity duration.
//For example, all inserts that happen in the first minute of an hour are stored
//in a key like "namespace:1min:0". All inserts from the second minute are stored
//in the "namespace:1min:60", and so on
TimeSeries.prototype._getRoundedTimestamp = function(timestampInSeconds, precision){
    return Math.floor(timestampInSeconds / precision) * precision;
};

//executes a callback by passing an array of data points
TimeSeries.prototype.fetch = function(granularityName, beginTimestamp, endTimestamp, onComplete){

    var granularity = this.granularities[granularityName];
    var begin = this._getRoundedTimestamp(beginTimestamp, granularity.duration);
    var end = this._getRoundedTimestamp(endTimestamp, granularity.duration);
    var keys = [];

    //iterate over all the timestamps in the specified range and save their values
    //in the "keys" variable
    for(var timestamp = begin; timestamp <= end; timestamp += granularity.duration){
        var key = this._getKeyName(granularity, timestamp);
        keys.push(key);
    }

    //the MGET command
    this.client.mget(keys, function(err, replies){
        var results = [];
        //iterate over all replies
        for(var i = 0; i < replies.length; i++){
            var timestamp = beginTimestamp + i * granularity.duration;
            //convert value to an integer
            var value = parseInt(replies[i], 10) || 0;
            //save timestamp and value in the "results" variable
            results.push({timestamp: timestamp, value:value});
        }
        //execute callback passing the variables "granularityName" and "results"
        onComplete(granularityName, results);
    });

};

//make a function available as a module in Node.js
exports.TimeSeries = TimeSeries;

//time series constructor. Requires a redis client and a namespace

function TimeSeries(client, namespace){

this.namespace = namespace;

this.client = client;

//granularity names and their equivalents in seconds

this.units = {

second:1,

minute: 60,

hour: 60 * 60,

day: 24 * 60 * 60

};

//each granularity has a name, TTL(time to live) and a duration.

//the null ttl present on 1dat meands that this ttl never expires

this.granularities = {

'1sec': {name: '1sec', ttl: this.units.hour * 2, duration: this.units.second},

'1min': {name: '1min', ttl: this.units.day * 7, duration: this.units.minute},

'1hour': {name: '1hour', ttl: this.units.day * 60, duration: this.units.hour},

'1day': {name: '1day', ttl: null, duration: this.units.day}

};

}

//insert a particular price at a given timestamp

TimeSeries.prototype.insert = function(timestampInSeconds, price){

//iterate over all franularities

for (var granularityName in this.granularities){

var granularity = this.granularities[granularityName];

//get a key name in the format "napespace:granularity:timestamp"

//for ex.: "google:1sec:12"

var key = this._getKeyName(granularity, timestampInSeconds);

//execute the SET command

this.client.set(key, price);

//the EXPIRE command

//pass the key and the ttl

//this command deletes a redis key automatically after a given number

//of seconds

if(granularity.ttl !== null){

this.client.expire(key, granularity.ttl);

}

};

//returns a key based on granularitu and timestamp

TimeSeries.prototype._getKeyName = function(granularity, timestampInSeconds){

var roundedTimestamp = this._getRoundedTimestamp(timestampInSeconds, granularity.duration);

return [this.namespace, granularity.name, roundedTimestamp].join(':');

};

//returns a normalized timestamp by granularity duration.

//For example, all inserts that happen in the first minute of an hour are stored

//in a key like "namespace:1min:0". All inserts from the second minute are stored

//in the "namespace:1min:60", and so on

TimeSeries.prototype._getRoundedTimestamp = function(timestampInSeconds, precision){

return Math.floor(timestampInSeconds / precision) * precision;

};

//executes a callback by passing an array of data points

TimeSeries.prototype.fetch = function(granularityName, beginTimestamp, endTimestamp, onComplete){

var granularity = this.granularities[granularityName];

var begin = this._getRoundedTimestamp(beginTimestamp, granularity.duration);

var end = this._getRoundedTimestamp(endTimestamp, granularity.duration);

var keys = [];

//iterate over all the timestamps in the specified range and save their values

//in the "keys" variable

for(var timestamp = begin; timestamp <= end; timestamp += granularity.duration){

var key = this._getKeyName(granularity, timestamp);

keys.push(key);

}

//the MGET command

this.client.mget(keys, function(err, replies){

var results = [];

//iterate over all replies

for(var i = 0; i < replies.length; i++){

var timestamp = beginTimestamp + i * granularity.duration;

//convert value to an integer

var value = parseInt(replies[i], 10) || 0;

//save timestamp and value in the "results" variable

results.push({timestamp: timestamp, value:value});

}

//execute callback passing the variables "granularityName" and "results"

onComplete(granularityName, results);

});

};

//make a function available as a module in Node.js

exports.TimeSeries = TimeSeries;

Now create a file called using-stock-timeseries.js, which will illustrate how to use our library. This file inserts stock quotes for a TimeSeries called “GAZPROM”, and then fetches values from a different granularities. Before inserting data, we remove all existing keys.

var redis = require("redis");
var client = redis.createClient();

//the FLUSHALL command
//removes all of the data from Redis
client.flushall();

var timeseries = require("./stock-timeseries");
//create a TimeSeries object passing the redis client and the "GAZPROM" namespace
//as an argument
var StocksGazprom = new timeseries.TimeSeries(client, "GAZPROM");

//this timestamp value was chosen to make it easier to read the output
var beginTimestamp = 0;

//execute the insert function
StocksGazprom.insert(beginTimestamp, 10);
//execute the insert function, passing a timestamp that is 1 second after "beginTimestamp"
StocksGazprom.insert(beginTimestamp + 1, 11);
StocksGazprom.insert(beginTimestamp + 2, 12);
StocksGazprom.insert(beginTimestamp + 3, 13);
StocksGazprom.insert(beginTimestamp + 4, 14);

//callback for displaying the output of the "fetch" function
function displayResults(granularityName, results){
    console.log("Results from", granularityName,":");
    console.log("Timestamp   | Value");
    console.log("----------  | ------");
    for(var i = 0; i < results.length; i++){
        console.log('\t' + results[i].timestamp + '\t' + results[i].value);
    }
    console.log();
}

//retrieve an interval of 5 seconds
StocksGazprom.fetch("1sec", beginTimestamp, beginTimestamp + 4, displayResults);
//retrieve an interval of 5 minutes
StocksGazprom.fetch("1min", beginTimestamp, beginTimestamp + 4, displayResults);

client.quit();

var redis = require("redis");

var client = redis.createClient();

//the FLUSHALL command

//removes all of the data from Redis

client.flushall();

var timeseries = require("./stock-timeseries");

//create a TimeSeries object passing the redis client and the "GAZPROM" namespace

//as an argument

var StocksGazprom = new timeseries.TimeSeries(client, "GAZPROM");

//this timestamp value was chosen to make it easier to read the output

var beginTimestamp = 0;

//execute the insert function

StocksGazprom.insert(beginTimestamp, 10);

//execute the insert function, passing a timestamp that is 1 second after "beginTimestamp"

StocksGazprom.insert(beginTimestamp + 1, 11);

StocksGazprom.insert(beginTimestamp + 2, 12);

StocksGazprom.insert(beginTimestamp + 3, 13);

StocksGazprom.insert(beginTimestamp + 4, 14);

//callback for displaying the output of the "fetch" function

function displayResults(granularityName, results){

console.log("Results from", granularityName,":");

console.log("Timestamp | Value");

console.log("---------- | ------");

for(var i = 0; i < results.length; i++){

console.log('\t' + results[i].timestamp + '\t' + results[i].value);

}

console.log();

}

//retrieve an interval of 5 seconds

StocksGazprom.fetch("1sec", beginTimestamp, beginTimestamp + 4, displayResults);

//retrieve an interval of 5 minutes

StocksGazprom.fetch("1min", beginTimestamp, beginTimestamp + 4, displayResults);

client.quit();

Now run your Redis server via redis-server. Then run node using-stock-timeseries.js. You should see the following output:

That’s all for today

Comprehensive guide to Redis. Part 3.

Leave a Reply Cancel reply