ITPub博客

首页 > 数据库 > NoSQL > 使用Spark和Cassandra进行数据处理(一)

使用Spark和Cassandra进行数据处理(一)

NoSQL 作者:liuya1985liuya 时间:2013-10-20 00:18:03 0 删除 编辑

最近在研究将Spark架设到Cassandra之上。发现这方面的信息比较少,在学习的过程中也遇到了不少问题,因此在此记录下,也和大家分享。此例为最经典的WordCount示例。

首先我先说下我所使用的各种环境和版本。由于Spark和Cassandra更新较快,如果之后版本有异可能运行不能成功需要一些微调。

暂时使用的是Windows 7, 之后会转到Linux平台,但是这个影响不大。使用的是Scala2.9.3,Spark 0.8, Cassandra 1.2.10,sbt 0.13.0,Java 7。

 

首先需要我们自己生成下Spark的jar包。这个需要我们运行sbt命令来得到。转到Spark所在的目录然后运行

sbtsbt assembly(如果是Linux的话,应该是sbt/sbt assembly)。运行结束后,可以在sparkassembly argetscala-2.9.3下面发现一个名字类似于spark-assembly-0.8.0-*.jar的包。我们需要将这个包加入到编译路径中。

 

其次我们需要在Cassandra中插入一些数据。首先需要在命令行中运行cassandra/bin下面的cassandra命令来开启服务。然后运行cassandra-cli,这样我们就能够输入我们需要的数据。本文结尾可以找到此例使用的数据示例。

 

然后我们就可以开始了。

Scala代码 
  1. val sc new SparkContext("local[3]""casDemo" 

 新建一个Spark上下文对象,第一个参数是将要连接到的Cluster地址,这里我们仅仅使用localhost来运行,所以可以简单设置为local[*],*为1,2,3之类的数字。第二个只是一个显示参数,可以随意。

Scala代码 
  1. val job new Job()  

 新建一个Hadoop job。因为Spark是没有提供直接的API访问Cassandra,但是Spark是建于Hadoop之上,Cassandra提供了访问Hadoop的接口所以我们需要先创建一个Hadoop的job来连接它两。

Scala代码 
  1. job.setInputFormatClass(classOf[ColumnFamilyInputFormat])  

 设置input的类,这个没有什么其他可选项,这是Cassandra默认的jar包中提供的接口。

 

Scala代码 
  1. ConfigHelper.setInputInitialAddress(job.getConfiguration(), "localhost" 
  2. ConfigHelper.setInputRpcPort(job.getConfiguration(), "9160" 
  3. ConfigHelper.setInputColumnFamily(job.getConfiguration(), "casDemo""Words" 
  4. ConfigHelper.setInputPartitioner(job.getConfiguration(), "Murmur3Partitioner" 

通过Cassandra提供的静态类ConfigHelper来设置相对应的一些参数。“casDemo"是这个例子使用的keyspace,words是column family。9160是Cassandra默认使用的端口。最后一个是设置多台机器运行时使用的Partitioner hash算法。有三个值可以选择,分别是org.apache.cassandra.dht.Murmur3Partitioner,org.apache.cassandra.dht.RandomPartitioner和ByteOrderedPartitioner。第三个已经不推荐使用了。在这里我们使用第一个。

 

此外需要说明的是,还有其他参数可以设置比如slice predicate之类的,这里略过了,仅仅介绍了最简单的设置。

 

然后我们就能够去创建Spark独有的RDD对象了,并使用它来完成我们要的工作。

Scala代码 
  1. val casRdd sc.newAPIHadoopRDD(job.getConfiguration(),  
  2.             classOf[ColumnFamilyInputFormat],  
  3.             classOf[ByteBuffer],//key  
  4.             classOf[SortedMap[ByteBuffer, IColumn]]) //value  

 可以看到这里创建了一个RDD对象,第一个参数就是将之间我们配置好的参数,第二个就是之前提到的Cassandra提供的接口,第三和第四个参数其他没有其他的可选项,这两个参数是被ColumnFamilyInputFormat所限制的。这个其实大家看SparkAPI就能了解到,这里不多说。

 

Scala代码 
  1. val paraRdd casRdd flatMap  
  2.     case (key, value) =>  
  3.         value.filter(v =>  
  4.             ByteBufferUtil.string(v._1).compareTo("paragraph"== 0  
  5.         }).map(v => ByteBufferUtil.string(v._2.value()))  
  6.      
  7.  

这里就是运行mapper方法了,就和hadoop中的mapper一个概念。需要说明的是,这里的key就是一个ByteBuffer对象,value就是一个SortedMap[ByteBuffer, IColumn],有没有觉得很熟悉,是的,这两个就是之前创建RDD设置的最后两个参数。这里我做的就是过滤掉colomn name不等于paragraph的,然后将colomn是paragraph的值由Icolumn变成ByteBuffer类型。

 

Scala代码 
  1. val counts paraRdd.flatMap(p => p.split(")).map(word => (word, 1)).reduceByKey(_ _)  

最后就是reduce方法了。 先用空格将段落打散成单词,然后将单个单词word转化成元组(word, 1)。最后统计每个word出现次数。

 

这样我们就完成了简单的WordCount方法。

我们可以通过以下代码打印出结果来观看。

Scala代码 
  1. counts.collect() foreach  
  2.     case (word, count) => println(word ":" count)  
  3.  

 

这里介绍了通过Spark读取Cassandra数据并进行处理。下一节,会介绍写入数据。


此文与http://cjcrobin.iteye.com/blog/1955984同步.可以在上面下载到此例源码。

<!-- 正文结束 -->

来自 “ ITPUB博客 ” ,链接:http://blog.itpub.net/8750110/viewspace-1120611/,如需转载,请注明出处,否则将追究法律责任。

上一篇: oralce 常见小问题
下一篇: 没有了~
请登录后发表评论 登录
全部评论

注册时间:2007-12-13

最新文章