BulkLoad To the Cassandra with the Hadoop

BulkOutputFormat

To bulkload the data to the Cassandra using Hadoop Cassandra introduces new OutputFormat that is BulkOutputFormat. Cassandra has implemented it in such a way that each map or reduce(depends on implementation) will generate sstables with data provided and then stream them to Cassandra with sstableloader. Don’t worry you need not know about all this implementation details to use BulkOutputFormat, all you need to know is some job configuration  and basic thrift call to create columns and mutations.

1. Initial Setup to write Hadoop job with BulkOutputFormat
2. Cassandra Related configuration
3. Using BulkOutputFormat

1)Initial Setup to write Hadoop job with BulkOutputFormat

1. To start with the development you need to have all the jars from Cassandra-1.1.x into classpath.
2. All the Hadoop related jars should be in classpath.
3. For execution of this job you should have all the Cassandra jars in classpath of Hadoop.

2)Cassandra Related configuration

We have to set following properties to Hadoop configuration Object.

      Keyspace name and ColumnFamily name:
      Name of the keyspace and column family for which we are creating sstables and streaming to Cassandra.

Use ConfigHelper class from Cassandra to set both these properties to Configuraion object of hadoop.

e.g

ConfigHelper.setOutputColumnFamily(conf,“keyspace_name”, columnFamily_name”);

      ThriftPort :

port for Thrift to listen for clients on

ConfigHelper.setOutputRpcPort(conf, “9160”);

(Note that you have to provide it as string)

    Seed node:

At least one node of the cluster you want to load data in should be configured as seed.

ConfigHelper.setOutputInitialAddress(conf, “10.10.10.10”);

(Note- ip address also have to specify as string)

     Partitioner:

 The partitioner is responsible for distributing rows (by key) across  nodes in the cluster.  Any  Partitioner may be used, including your  own as long as it is on the classpath.

ConfigHelper.setOutputPartitioner(conf,“org.apache.cassandra.dht.RandomPartitioner”);

    Buffer Size:
      SSTableSimpleUnsortedWriter buffers whatever input it gets in memory and “flush” everything in one sstable once the buffer size is reached.   That is, the resulting sstables will be approximately of size equal to buffer in MB.

You have to set this directly in hadoop configuration object

conf.set(“mapreduce.output.bulkoutputformat.buffersize

”, “64”);

(Note-  Buffer size is specified in string)

    OutputFormat class:

Set output format class for the job as BulkOutPutFormat

job.setOutputFormatClass(BulkOutputFormat.class);

So you have done with the basic configuration needed for the BulkOutputFormat.
Sample code:

Configuration conf = new Configuration();

ConfigHelper.setOutputColumnFamily(conf, “keyspace_name”, “columnFamily_name”);

ConfigHelper.setOutputRpcPort(conf, “9160”);

ConfigHelper.setOutputInitialAddress(conf, “10.10.10.10”);

ConfigHelper.setOutputPartitioner(conf,“org.apache.cassandra.dht.RandomPartitioner”);

conf.set(“mapreduce.output.bulkoutputformat.buffersize”, “64”);

Job job ;
job = new Job(conf);

job.setOutputFormatClass(BulkOutputFormat.class);

/*Rest of your job related configuration like input format , input path and all will go here*/

3)Using  BulkOutputFormat

If you look at the declaration of BulkOutputFormat  It looks like following

public class BulkOutputFormat extends OutputFormat<ByteBuffer,List<Mutation>>

That means you should have output key as Bytebuffer and Output value as list of Mutation.
It means in map or reduce you have to process  data to produce the rowkey as bytebuffer and columns corresponding to this row key as list of mutations. And when Hadoop calls

 context.write(ByteBuffer,List<Mutation>)

Sstables will be generated into map tmp directory and once the  record.close() is invoked it will start loading sstables to Cassandra.

You don’t need to know all this implementation details , all we need to know is create row-key of type ByteBuffer and columns in the form of List<Mutation> , and write them using context.write().Everything else will be taken care by Hadoop and Cassandra.

Creating row-key and list of mutations

ByteBuffer rowkey = org.apache.cassandra.utils.ByteBufferUtil.bytes(“rowkey”);

Column column = new Column();

            column.name = name;

            column.value = value;

            column.timestamp = timestamp;

            column.ttl = ttl;

List<Mutation> mutationList;
Mutation mutation = new Mutation();

            mutation.column_or_supercolumn = new ColumnOrSuperColumn();

            mutation.column_or_supercolumn.column = column;

            mutationList.add(mutation);

context.write(rowkey, mutationList);

IF you want to write the supercolumn

Create a list of all columns those are related to supercolumn.Add these columns to column list.Create supercolumn using this columnlist , create mutation out of this supercolumn and add this mutation to mutation list.

Example:
 

List<Column> columnList;

Column column1 = new Column();

            column1.name = name;

            column1.value = value;

            column1.timestamp = timestamp;

            column1.ttl = ttl;

Column column2 = new Column();

            column2.name = name;

            column2.value = value;

            column2.timestamp = timestamp;

            column2.ttl = ttl;

columnList.add(column1);
columnList.add(column2);

List<Mutation> mutationList;
SuperColumn superColumn = new SuperColumn(supercolumn, columnList);

            Mutation mutation = new Mutation();

            mutation.column_or_supercolumn = new ColumnOrSuperColumn();

            mutation.column_or_supercolumn.super_column = superColumn;

            mutationList.add(mutation);

context.write(rowkey, mutationList);

4)Tuning  generation and loading for optimize performance of the Cassandra

•  Deciding on an appropriate value of the Buffer is very important part of the sstable generation.
•  As setting this buffer size low results into faster generation and loading, but it puts load on cassndra and one should not set it to smaller value.
• With small buffer size the large no of small size sstables will be generated which when loaded into the Cassandra takes too much time to complete compaction and puts load on Cassandra cluster.
• Also you can turn minor compaction off while loading data into Cassandra for faster loading. (You can do this with nodetool setcompactionthreshoul)