The nodes that have Hadoop on them need to have virtual nodes disabled. So, if you have a Cassandra cluster, you may want to split it into two data centers. You will need to configure a proper endpoint snitch (such as PropertyFileSnitch) to separate the two data centers. The data centers need not actually be in different data centers. To disable the virtual node, change the following in cassandra.yaml:

If you are trying to divide an existing cluster into two parts, one for data storage (the normal vnode-enabled data center for data storage) and another for analytics (the ones that have vnode disabled and stays along with Hadoop), make sure, in analytics data center, you decommission those nodes then clean data from data directories and restart them after making the previously mentioned changes. This is the same as adding new nodes to the cluster. Disabling vnode in an existing running cluster is not advised.

The ColumnFamilyInputFormat class is an implementation of org.apache.hadoop.mapred.InputFormat (or mapreduce in newer the API). So, its implementation is dictated by the InputFormat class specifications. Hadoop uses this class to get data for the MapReduce tasks. It describes how to read data from column families into the Mapper instances.

The other job of ColumnFamilyInputFormat (or any implementation of InputFormat) is to fragment input data into small chunks that get fed to map tasks. Cassandra has ColumnInputSplit for this purpose. One can configure the number of rows per InputSplit via ConfigHelper.setInputSplitSize. However, there is a caveat. It uses multiple get_slice_range queries for each InputSplit data, so, as Cassandra documentation says, a smaller value will build up call overhead; on the other hand, too large a value may cause out-of-memory issues. Larger values are better for performance, so if you are planning to play with this parameter do some calculation based on median column size to avoid memory overflow. Trial and error can be handy. The default split size is 64 x 1024 rows.

The OutputFormat class is the mechanism of writing the result from MapReduce to a permanent (usually) storage. Cassandra implements Hadoop's OutputFormat, that is, ColumnFamilyOutputFormat. It enables Hadoop to write the result from the reduce task as column family rows. It is implemented such that the results are written, to the column family, in batches. This is a performance improvement, and this mechanism is called lazy write-back caching.

CqlOutputFormat and CqlInputFormats are Hadoop-specific output and input formats for reducer and mapper tasks, respectively. Functioning similar to ColumnFamilyOutputFormat and ColumnFamilyInputFormat, they provide the ability to access CQL rows and variable binding.

CqlInputFormat requires the keyspace and table name to be specified. You can use ConfigHelper class to set up this and other configurations. A couple of things that you should set are input split size via ConfigHelper.setInputSplitSize, which defaults to 64,000 rows. The number of CQL rows per page via CqlConfigHelper.setInputCqlPageRowSize defaults to 1,000 rows per page. It is a good idea to have CQL rows per page as big as your machine can support without causing memory issues. This will help reducing network overhead. Initial input address and partitioner may be required to mention. To do so, use the ConfigHelper.setInputInitialAddress and ConfigHelper.setInputPartitioner methods.

The CqlOutputFormat allows the reducer task to write keys and values to the specified CQL table. You need to set the output table, output initial address, and output partitioner via ConfigHelper and the CQL that updates the output table via CqlConfigHelper.setOutputCql.

The ConfigHelper class is a gateway to configure Cassandra-specific settings for Hadoop. It is a pretty plain utility class that validates the settings passed and sets into Hadoop's org.apache.hadoop.conf.Configuration instance for the job. This configuration is made available to the Mapper and the Reducer.

The ConfigHelper class saves developers from inputing the wrong property name because all the properties are set using a method; any typo can appear at compile time. It may be worth looking at JavaDoc for ConfigHelper. Here are some of the commonly used methods:

Since version 1.1, Cassandra added support to wide row column families, bulk loading, and secondary indexes.

ConfigHelper.setInputColumnFamily(
conf,
keyspace,
inCF,
true// SET WIDEROW = TRUE
);

Job job = new Job(conf);
job.setOutputFormatClass(BulkOutputFormat.class);

IndexExpressionelectronicItems =
    newIndexExpression(
    ByteBufferUtil.bytes("item_category"),
    IndexOperator.EQ,
    ByteBufferUtil.bytes("electronics")
      );

IndexExpressionsoldAfter2012 =
    newIndexExpression(
    ByteBufferUtil.bytes("sell_year"),
            IndexOperator.GT,
    ByteBufferUtil.bytes(2012)
  );

ConfigHelper.setInputRange(conf, Arrays.asList(electronicItems, soldAfter2012));