Recently in NoSQL Category

When trying to evaluate NoSQL databases, its usually better to try them out. While trying them out, its better to use them with multiple node configurations instead of running single node. Such as clusters in Riak or Replica-set in mongodb maybe even a sharded setup. On our project we evaluated a 10 node Riak cluster so that we could experiment with N, R and W values and decide which values where optimal for us. In Riak here is what N, R and W mean

N = Number of Riak nodes to which data will be replicated

R = Number of Riak nodes which have to return results for the read to be considered successful

W = Number of Riak nodes which have to return a write success before the write is considered successful

These N,R and W settings provide us the ability to tune our CAP requirements, thus they need to be carefully considered when architecting the system. What better way is there to test the assumptions we make than trying the assumptions out with some code and running Riak nodes. To experiment with our assumptions, we built a script that will download Riak 1.3.1 and create 10 nodes with different ports for pb_port, http port, handoff_port by changing them in app.config and -name in vm.args file for each node.

The first node is used as a master node to which all other nodes join after they are started using the riak-admin cluster join master_node_name@127.0.0.1 command to join the cluster. When all the nodes have been started, we can look at the cluster plan or configuration using riak-admin cluster plan and then commit the cluster plan using riak-admin cluster commit. This commits the cluster changes and makes all the nodes part of a cluster, we can view the status of the cluster using riak-admin status or see the nodes of the cluster using riak-admin status | grep 'ring_members'.

The shell script to set up 10 node Riak cluster.

#!/bin/bash -e
working_dir=`pwd`
master_node_id='1'
node_directory='riak-node'
node_name_prefix='node'
master_node_name=$node_name_prefix$master_node_id
echo "Checking if Riak is downloaded"
if [ ! -f "riak-1.3.*.tar" ]; then
  rm -rf riak-1.3.*.tar
  wget http://s3.amazonaws.com/downloads.basho.com/riak/1.3/1.3.1/osx/10.6/riak-1.3.1-osx-x86_64.tar.gz
fi
echo "Unzip and Set up Riak"
gunzip riak-1.3.1-osx-x86_64.tar.gz 
tar -xf riak-1.3.1-osx-x86_64.tar
echo "Setting up 10 nodes"
for i in {1..10}
do
  protocolbuffer_port=`expr 8000 + $i`
  http_port=`expr 8100 + $i`
  handoff_port=`expr 8200 + $i`
  cd $working_dir
  nodeid=$node_directory$i
  cp -r riak-1.3.1 $nodeid
  sed -e s/8087/$protocolbuffer_port/g -i '' $nodeid/etc/app.config
  sed -e s/8098/$http_port/g -i '' $nodeid/etc/app.config
  sed -e s/8099/$handoff_port/g -i '' $nodeid/etc/app.config
  sed -e s/riak@/$node_name_prefix$i@/g -i '' $nodeid/etc/vm.args
  cd $nodeid/bin
  ./riak start
  if [ $i -ne $master_node_id ]; then
    ./riak-admin cluster join $master_node_name@127.0.0.1
  fi	
done
cd $working_dir/$node_directory$master_node_id/bin/
./riak-admin cluster plan
./riak-admin cluster commit
./riak-admin status | grep 'node.@'
echo "10 node Riak cluster setup"

*These shell scripts works on my mac and has not been tested on anything other than OSX-10.7+

Once experimenting, prototyping with the Riak cluster is done, we can gracefully shutdown the riak cluster and clean up all the folders using this shell script

#!/bin/bash -e
working_dir=`pwd`
master_node_id='1'
node_directory='riak-node'
echo "Cleaning up riak cluster"
for i in {1..10}
do
  echo "Shutdown node:"$i
  nodeid=$node_directory$i
  cd $working_dir
  cd $nodeid/bin
  ./riak stop
done
cd $working_dir
./riak-node1/erts-5.9.1/bin/epmd -kill
rm -rf riak-node*
rm *.tar
rm -rf riak-1.3.1

These scripts and approach can be modified to work across platforms

There are multiple ways to take backups of mongodb is different configuraitions, one of the configuration that I have been involved recently is replica-sets. When mongodb is running in replica-set configuration, there is a single primary node and multiple secondary nodes. To take backup of the replica-set we can either do a mongodump of one of the nodes or shutdown one of the secondary nodes and take file copies, since in a replica-set all nodes have the same data (except arbiter). Lets see how we could deal with mongodump method of taking backup.

Sync the primary node so that all writes are flused to disk and lock the database for writes, doing a fsync allows for all writes to be persisted to the disk.

use admin
db.fsyncLock()

now we can issue the mongodump command, there are many more options for mongodump that can be changed, using defaults here

mongodump -h node_name  --out /data/backups/backup_file_name

once the mongodump command is done we can unlock the database so that writes can be issued.

use admin
db.fsyncUnlock()

The downside of this approach is that the primary is not available for writes, but reads are fine. If by any chance there is write issued, all reads are also blocked after that, which is pretty drastic. The other option is to operate on one of the secondaries, this allows us to keep the primary available for write and read, the secondary can be used for backup purposes. Which node is PRIMARY or SECONDARY can be dynamically determined by running some javascript on the command line

#!/usr/bin/env ruby
require 'rubygems'
require 'json'
mongo_nodes = JSON.parse `mongo node_name --quiet 
              --eval "printjson(rs.status().members.map(
                        function(m) { return 
                       {'name':m.name, 'stateStr':m.stateStr} }))"`
primary_node = mongo_nodes.detect { 
                   |member| member['stateStr'] == 'PRIMARY'
                    }

This dynamic script allows us to find the node that we want to take backup from, either the primary or secondary.

Replica sets is a feature of MongoDB for Automatic Failover, in this setup there is a primary server and the rest are secondary servers. If the primary server goes down, the rest of the secondary servers choose a new primary via an election process, each server can also be assigned number of votes, so that you can decide the next primary based on data-center location, machine properties etc, you can also start mongo database processes that act only as election tie-breakers these are known as arbiters, these arbiters will never have data, but just act as agents that break the tie.

All operations are directed at the primary server, the primary server writes the operations to its operation log (also known as opslog), the secondary servers get updates from the primary server. The data is written to the primary server and later replicated to the other secondary servers, so when the write happens at the primary and before the write is replicated to the secondary servers, if the primary server goes down you will loose the data that was written to the primary but never replicated to the secondary servers, you can get around this by specifying how many servers should have the data, before the write is considered good

db.runCommand( { getlasterror : 1 , w : 3 } )

in the above command, you are saying that the write to the database is considered good, only if the write has been propagated to at least 3 servers, off course doing this for every write is going to be very expensive, so you should batch all your writes for a user action and then issue getlasterror

This is how you start the the mongod servers, in a replica set, the can run on any machine any port, as long as they can talk to each other over the network and all of them have the same "--replSet" parameter, in the example below its "prod"

mongod --replSet prod --port 27017 --dbpath /data/node1 
mongod --replSet prod --port 27027 --dbpath /data/node2 
mongod --replSet prod --port 27037 --dbpath /data/node3 

Once the three servers are up, you have to create a replica configuration as shown below, if you use localhost as a server name, then all the members of the replica set have to be on localhost, if the mongo servers are on different servers, you should use distinct machine names and not localhost for anyone of them, once the replica config is defined, you then initiate the replica using the configuration as shown below

replica_config = {_id: 'prod', members: [
                          {_id: 0, host: 'localhost:27017'},
                          {_id: 1, host: 'localhost:27027'},
                          {_id: 2, host: 'localhost:27037'}]}
#Now initiate the replica_config
rs.initiate(replica_config);

When you are connecting to a replica set, you have to connect to atleast one server which is alive, using the ruby driver you can connect to more than one server using the "multi" method, one part you should be careful about is, lets say you define all the servers in the replica set as your connection string, but one of the members of the replica set is down, you will get connection failures, so the best thing to do is give members of the replica set that are up and the drivers will discover the other servers when they come online or go offline. Here is a sample ruby program to find a doc in a loop.

#!/usr/bin/env ruby
require 'mongo'
begin
  @connection = Mongo::Connection.multi([
                                       ['localhost',27017],
                                       ['localhost',27027],
                                       ['localhost',27037]])
  @collection = @connection.db("sales").collection("products")
  product = { "name" => "Refactoring", 
              "code" => "023XX3",
              "type" => "book", 
              "in_stock" => 100}
  @collection.insert(product)
  100.times do
    sleep 0.5
    begin
  	  product = @collection.find_one "code" => "023XX3"
  	  puts "Found Book: "+product["name"]
  	rescue Exception => e
    	puts e.message
    	next
  	end
  end
end

While the ruby program is running, you can kill the current primary and you will see that the program gets connection exceptions, while the replica set is figuring out the next master, once the next master is picked, the program starts going about its way finding the same data from the newly elected primary, here is a screen cast of the replica sets in action. Replica Sets screencast

In the No-SQL land schema-less is a power full feature that is advertised a lot, schema-less basically means you don't have to worry about column names and table names in a traditional sense, if you want to change the column name you just start saving the data using the new column name Lets say you have a document database like mongoDB and you have JSON document as shown below.

{  "_id":"4bc9157e201f254d204226bf",
   "FIRST_NAME":"JOHN",
   "MIDDLE_NAME":"D",
   "LAST_NAME":"DOE",
   "CREATED":"2010-10-12"
}

You have some corresponding code to read the documents from the database and lets say you lots of data in the database in the order of millions of documents. If you want to change the name of some attributes or columns at this point and the new JSON would look like

{  "_id":"4bc9157e201f254d204226bf",
   "first_name":"JOHN",
   "middle_name":"D",
   "last_name":"DOE",
   "created":"2010-10-12"
}

You will have to either change every document in the database to match the new attribute names or you have to make sure you code can handle both types of attribute names like

   first_name = doc["first_name"] 
   first_name = doc["FIRST_NAME"] unless !first_name.nil? 
   middle_name = doc["middle_name"]  
   middle_name = doc["MIDDLE_NAME"] unless !middle_name.nil?
   last_name = doc["last_name"]
   last_name = doc["LAST_NAME"] unless !last_name.nil?

This attribute name change also affects the indexes created on mongoDB, since the attribute name change is not across all the documents, an Index created on

   db.people.ensureIndex({first_name:1})

will not index documents where the attribute name is FIRST_NAME, so you have to create another index for this new attribute name

   db.people.ensureIndex({FIRST_NAME:1})

As you can see this gets really complicated if you do multiple refactorings, over a period of time. So when you hear schema less make sure you understand the ramifications of refactoring the attribute names at will and its effect on the code base and the database.