概要

前提

假设当前已经安装好如下软件

• jdk
• sbt
• git
• scala

安装cassandra

以archlinux为例，使用如下指令来安装cassandra

yaourt -S cassandra

启动cassandra

cassandra -f

创建keyspace和table, 运行/usr/bin/cqlsh进入cql console，然后执行下述语句创建keyspace和table

CREATE KEYSPACE test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1 };
CREATE TABLE test.kv(key text PRIMARY KEY, value int);

 添加记录，继续使用cql console

INSERT INTO test.kv(key, value) VALUES ('key1', 1);
INSERT INTO test.kv(key, value) VALUES ('key2', 2);

验证记录已经插入成功，执行如下cql

select * from test.kv;

下载编译spark-cassandra-connector

下载新的spark-cassandra-connector源码

git clone https://github.com/datastax/spark-cassandra-connector.git

编译

sbt package

这中间要等待比较长的时间，请保持足够的耐心

运行spark-shell

首先请确保cassandra已经正常安装和运行，如有问题请返回开始的章节安装cassandra

如何添加相应的library来支持spark-cassandra-connector，并没有一个明确的文档说明，折腾了一个下午，终于弄出了一个简的配置

bin/spark-shell --driver-class-path /root/working/spark-cassandra-connector/spark-cassandra-connector/target/scala-2.10/spark-cassandra-connector_2.10-1.1.0-SNAPSHOT.jar:
/root/.ivy2/cache/org.apache.cassandra/cassandra-thrift/jars/cassandra-thrift-2.0.9.jar:
/root/.ivy2/cache/org.apache.thrift/libthrift/jars/libthrift-0.9.1.jar:
/root/.ivy2/cache/org.apache.cassandra/cassandra-clientutil/jars/cassandra-clientutil-2.0.9.jar:
/root/.ivy2/cache/com.datastax.cassandra/cassandra-driver-core/jars/cassandra-driver-core-2.0.4.jar:
/root/.ivy2/cache/io.netty/netty/bundles/netty-3.9.0.Final.jar:
/root/.ivy2/cache/com.codahale.metrics/metrics-core/bundles/metrics-core-3.0.2.jar:
/root/.ivy2/cache/org.slf4j/slf4j-api/jars/slf4j-api-1.7.7.jar:
/root/.ivy2/cache/org.apache.commons/commons-lang3/jars/commons-lang3-3.3.2.jar:
/root/.ivy2/cache/org.joda/joda-convert/jars/joda-convert-1.2.jar:
/root/.ivy2/cache/joda-time/joda-time/jars/joda-time-2.3.jar:
/root/.ivy2/cache/org.apache.cassandra/cassandra-all/jars/cassandra-all-2.0.9.jar:
/root/.ivy2/cache/org.slf4j/slf4j-log4j12/jars/slf4j-log4j12-1.7.2.jar

上述指令假设spark-cassandra-connector的源码是下载在$HOME/working目录下，请根据自己的情况作适当修改

我是如何猜测到需要指定这些包依赖的呢？说白了，也很简单，就是执行以下指令，然后再查看相就的java进程中的运行参数

#运行spark-cassandra-connector测试集
sbt test
sbt it:test

当上述指令还在运行的时候，使用ps来查看java运行的参数，这样就反过来知道所需要的包依赖了

ps -ef|grep -i java

测试程序

由于spark-shell会默认创建sc,所以首先需要停止掉默认的sc，然后利用新的配置来创建可以连接到cassandra的sc,示例代码如下

sc.stop
import com.datastax.spark.connector._
import org.apache.spark._
val conf = new SparkConf()
conf.set("spark.cassandra.connection.host", "127.0.0.1")
val sc = new SparkContext("local[2]", "Cassandra Connector Test", conf)
val table = sc.cassandraTable("test", "kv")
table.count

如果一切正常会显示出如下结果

res3: Long = 2

小结

进入实战阶段，挑战会越来越多，保持足够的信心和耐心很重要。

本篇内容和实战一中的kafka cluster组织在一起的话，就会形成一个从前台到后台存储的完整处理链条。