Export (and import) Redis keys in a normalized way to Riak.

redis2riak.js

var redisScan = require('redisscan');
var redis     = require('redis');
var riak      = require('riakpbc');
var async     = require('async');
var args      = require('optimist')
    .usage('--import or --export Redis archives in Riak\nUsage: $0')
    .boolean('import')
    .boolean('export')
    .boolean('debug')
    .describe('import', "Import from Riak to Redis")
    .describe('export', "Export from Redis to Riak")
    .describe('debug', "Output debug logs to stdout")
    .default({'redis': 'localhost:6379', riak: 'localhost:8087', sep: '_!!_'})
    .demand(['bucket'])
    .check(function (args) {
        if (!(args.import || args.export)) {
           throw new Error('--import or --export is required');
        }
        if (args.import && args.export) {
           throw new Error('Either select --import or --export, not both');
        }
        return true;
    }).argv;


function redis2riak(opts) {
    redisScan({
        redis: opts.redis,
        each_callback: function (type, key, subkey, ln, value, cb) {
            if (args.debug) console.log(type + opts.sep + key + opts.sep + subkey + opts.sep + ln, '=', value)
            opts.riak.put({
                bucket: opts.bucket,
                key: type + opts.sep + key + opts.sep + subkey + opts.sep + ln,
                content: {
                    value: value,
                    content_type: 'text/plain'
                },
            }, cb);
        },
        done_callback: function (err) {
            console.log("done");
            opts.redis.quit();
            opts.riak.disconnect();
        }
    });
}

function riak2redis(opts) {
    var lists = {};
    opts.riak.getKeys({bucket: opts.bucket}, function (err, reply) {
        var keys = reply.keys;
        async.eachSeries(keys, function (key, acb) {
            opts.riak.get({bucket: opts.bucket, key: key}, function (err, reply) {
                var parts = key.split(opts.sep);
                var args;
                if (parts[0] === 'string') {
                    opts.redis.set(parts[1], reply, acb);
                } else if (parts[0] === 'hash') {
                    opts.redis.hset(parts[1], parts[2], reply, acb);
                } else if (parts[0] === 'set') {
                    opts.redis.sadd(parts[1], reply, acb);
                } else if (parts[0] === 'zset') {
                    opts.redis.zadd(parts[1], parts[2], reply, acb);
                } else if (parts[0] === 'list') {
                    //if we've not encountered this list, make an empty one of its size
                    if (!lists.hasOwnProperty(parts[1])) {
                        lists[parts[1]] = true;
                        args = [parts[1]];
                        var length = parseInt(parts[3]);
                        for (var idx = 0; idx < length; idx++) {
                            args.push('x');
                        }
                        args.push(function (err) {
                            opts.redis.lset(parts[1], parts[2], reply, acb);
                        });
                        opts.redis.lpush.apply(opts.redis, args);
                    } else {
                        opts.redis.lset(parts[1], parts[2], reply, acb);
                    }
                }
            });
            if (args.debug) console.log('key:', key)
        }, function (err) {
            console.log("done");
            opts.redis.quit();
            opts.riak.disconnect();
        });
    });
}

(function () {
    var riak_split = args.riak.split(':');
    var redis_split = args.redis.split(':');
    var opts = {
        redis: redis.createClient({host: redis_split[0], port: parseInt(redis_split[1])}),
        riak: riak.createClient({host: riak_split[0], port: parseInt(riak_split[1])}),
        bucket: args.bucket,
        sep: args.sep,
        debug: args.debug
    };
    if (args.import) {
        riak2redis(opts);
    } else {
        redis2riak(opts);
    }
})();