Cassandra-Map Reduce Over Cassandra

Following is the Map Reduce program that will read a column from Cassandra table and the column values to HDFS .

I have only written a Mapper class , because Reducer is not required .

Here is the Driver Class:

public class CassandraDriver {

    private static final String CASSANDRA_HOST = "192.168.213.1";

    private static final String CASSANDRA_PORT = "9160";

    private static final String KEYSPACE = "mykeyspace";

    private static final String COLUMN_FAMILY = "testhadoop1";

    private static final String CONF_COLUMN_NAME1 = "value1";

    private static final String CONF_COLUMN_NAME2 = "value2";

    public static void main(String[] args) throws IOException,

            InterruptedException, ClassNotFoundException {

        Configuration conf = new Configuration();

        Job job = new Job(conf, "Cassandra");

        job.setJarByClass(CassandraDriver.class);

        job.setMapperClass(CassandraMapper.class);

        job.setOutputKeyClass(NullWritable.class);

        job.setOutputValueClass(Text.class);

        job.setInputFormatClass(ColumnFamilyInputFormat.class);

        FileOutputFormat.setOutputPath(job, new Path(

                args[0]));

        ConfigHelper.setRangeBatchSize(job.getConfiguration(), 99);

        ConfigHelper.setInputRpcPort(job.getConfiguration(), CASSANDRA_PORT);

        ConfigHelper.setInputInitialAddress(job.getConfiguration(),

                CASSANDRA_HOST);

        ConfigHelper.setInputPartitioner(job.getConfiguration(),

                "RandomPartitioner");

        ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE,

                COLUMN_FAMILY);

        SlicePredicate predicate = new SlicePredicate().setColumn_names(Arrays

                .asList(ByteBufferUtil.bytes(CONF_COLUMN_NAME1),

                        ByteBufferUtil.bytes(CONF_COLUMN_NAME2)));

        ConfigHelper.setInputSlicePredicate(job.getConfiguration(), predicate);

        job.waitForCompletion(true);

    }

}

Here is the Cassandra Mapper:

public class CassandraMapper extends
        Mapper<ByteBuffer, SortedMap<ByteBuffer, IColumn>, NullWritable, Text> {
    private NullWritable mapkey = NullWritable.get();
    private Text word = new Text();

    public void map(ByteBuffer key, SortedMap<ByteBuffer, IColumn> columns,
            Context context) throws IOException, InterruptedException {
        for (IColumn column : columns.values()) {
        
          
               String  value = ByteBufferUtil.string(column.value());
           

            StringTokenizer itr = new StringTokenizer(value);
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                context.write(mapkey, word);
            }
        }
    }

}