整合Cassandra与Hadoop MapReduce

filer

　　看到这个标题，大家一定会问了。这个整合如何定义？
　　我个人认为，所谓的整合是指：我们可以编写MapReduce程序，从HDFS中读取数据然后插入到Cassandra中。也可以是直接从Cassandra中读取数据，然后进行相应的计算。

从HDFS中读取数据然后插入到Cassandra中
　　对于这种类型，我们可以按照以下几个步骤来操作。
　　1 将需要插入Cassandra的数据上传到HDFS中。
　　2 启动MapReduce程序。
　　这种类型的整合其实和Cassandra本身没有什么联系。我们只是运行普通的MapReduce程序，然后在Map或者Reduce端将计算好的数据插入到Cassandra中。仅此而已。

直接从Cassandra中读取数据，然后进行相应的计算
　　这个功能是在Cassandra0.6.x版本中添加上去的。其可以从Cassandra直接读取MapReduce需要的数据，实现对于Cassandra的全表扫描的功能。
　　操作步骤如下：
　　1 在MapReduce程序中指定使用的KeySpace，ColumnFamily，和SlicePredicate等和Cassandra相关的参数。（关于这些概念，可以参考《大话Cassandra数据模型》和《谈谈Cassandra的客户端》）
　　2 启动MapReduce程序。
　　这种类型的整合和从HDFS读取数据的整合相比，还是有许多不同的，主要有下面几点区别：
　　1 输入数据来源不同：前一种是从HDFS中读取输入数据，后一种是从Cassandra中直接读取数据。
　　2 Hadoop的版本不同：前一种可以使用任何版本的Hadoop，后一种只能使用Hadoop0.20.x

整合Hadoop0.19.x与Cassandra0.6.x
　　在Cassandra0.6.x中，默认实现的是与Hadoop0.20.x的整合，我们无法直接在Hadoop0.19.x中使用。
　　所以，要实现这个目标，我们第一步需要做的事情是，修改Cassandra的源代码，提供一个可以在Hadoop0.19.x中使用的功能。
　　想要进行这项测试，我们可以按照如下步骤来进行：
　　1 下载修改后的代码。
　　2 在MapReduce中指定如下内容（注意，这里的class使用的package都是com.alibaba.dw.cassandra.hadoop下面的）：





ConfigHelper.setColumnFamily(conf, Keyspace, MemberCF,
                "/home/admin/apache-cassandra-0.6.1/conf");
        SlicePredicate predicate = new SlicePredicate().setColumn_names(Arrays.asList("CITY"
                .getBytes(UTF8), "EMPLOYEES_COUNT".getBytes(UTF8)));
        ConfigHelper.setSlicePredicate(conf, predicate);
        ConfigHelper.setRangeBatchSize(conf, 512);
        ConfigHelper.setSuperColumn(conf, "MemberInfo");
　　3 确保每一台运行MapReduce的机器的指定目录与MapReduce程序中设定的storage-conf.xml文件路径一致。
　　4 运行MapReduce程序。

存在的问题与改进
　　在实际的使用中，我们会发现Map端会出现这样的错误信息：





java.lang.RuntimeException: TimedOutException()
    at com.alibaba.dw.cassandra.hadoop.ColumnFamilyRecordReader$RowIterator.maybeInit(ColumnFamilyRecordReader.java:125)
    at com.alibaba.dw.cassandra.hadoop.ColumnFamilyRecordReader$RowIterator.computeNext(ColumnFamilyRecordReader.java:164)
    at com.alibaba.dw.cassandra.hadoop.ColumnFamilyRecordReader$RowIterator.computeNext(ColumnFamilyRecordReader.java:1)
    at com.google.common.collect.AbstractIterator.tryToComputeNext(AbstractIterator.java:135)
    at com.google.common.collect.AbstractIterator.hasNext(AbstractIterator.java:130)
    at com.alibaba.dw.cassandra.hadoop.ColumnFamilyRecordReader.next(ColumnFamilyRecordReader.java:224)
    at com.alibaba.dw.cassandra.hadoop.ColumnFamilyRecordReader.next(ColumnFamilyRecordReader.java:1)
    at org.apache.hadoop.mapred.MapTask$TrackedRecordReader.moveToNext(MapTask.java:192)
    at org.apache.hadoop.mapred.MapTask$TrackedRecordReader.next(MapTask.java:176)
    at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:48)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:342)
    at org.apache.hadoop.mapred.Child.main(Child.java:158)
Caused by: TimedOutException()
    at org.apache.cassandra.thrift.Cassandra$get_range_slices_result.read(Cassandra.java:11015)
    at org.apache.cassandra.thrift.Cassandra$Client.recv_get_range_slices(Cassandra.java:623)
    at org.apache.cassandra.thrift.Cassandra$Client.get_range_slices(Cassandra.java:597)
    at com.alibaba.dw.cassandra.hadoop.ColumnFamilyRecordReader$RowIterator.maybeInit(ColumnFamilyRecordReader.java:108)
    ... 11 more
　　
　　引起这样的问题的原因就在于使用Thrift API从Cassandra读取数据失败了。
　　所以我们可以优化这段代码，提供想要的错误处理功能来提供程序的可用性。
　　另一个做法是修改Cassandra的配置，将RPCTimeout的时间调长。
　　
　　更多关于Cassandra的文章：http://www.iyunv.com/gpcuster/tag/Cassandra/