我们知道，Cassandra这种NoSQL数据库，天生无法执行join的操作。 但是如果你手上刚好有一个Spark集群，那么就方便很多了。我们可以在Spark SQL之中进行join的操作。
本文基于Spark 2.x 进行操作。2.0以后，我们不再需要单独的定义JavaSparkContext / SparkConf 等对象，只需直接定义一个SparkSession即可。同时我们可以统一使用Dataset来对数据进行操作，在易用性、性能上面都很不错。
下面是链接Spark与Cassandra的相关代码:
|
1 2 3 4 5 6 7 8 9 10 11 12 |
public static SparkSession getSession(String taskName) { SparkSession ss = SparkSession.builder() .master("spark://10.206.132.101:7077") .appName(taskName) .config("spark.cassandra.connection.host", "10.206.132.14,10.206.132.77") .config("spark.driver.memory", "5G") .config("spark.executor.memory", "18G") .config("spark.cores.max", "72") .getOrCreate(); ss.sparkContext().addJar("target/sparkstat-0.0.1-SNAPSHOT.jar"); return ss; } |
注意上面的第10行代码，我们最后需要打包成一个fat-jar，因此需要特别的addJar
当我们拿到SparkSession 之后，后面的事情就很简单了：
首先使用Cassandra的Spark-connector加载数据：
|
1 2 3 4 5 6 7 |
Map<String, String> options = Maps.newHashMap(); options.put("keyspace", KEYSPACE); options.put("table", TB_USER_NORMAL_INFO); SparkSession ss = ... Dataset<Row> ds = ss.read().format("org.apache.spark.sql.cassandra").options(options).load(); |
在Java API之中，统一使用Dataset啦，其API与DataFrame兼容，因为DataFrame只是Dataset的别名。
拿到数据之中，就可以使用SQL进行操作了：
|
1 2 3 4 5 |
// 首先注册一个表 ds.createOrReplaceTempView("localtest"); // 然后调用sqlContext使用SQL ds.sqlContext().sql("select * from localtest").show(); |
支持了普通的SQL，join操作自然也会支持了。 在此暂时略过~
文章评论