使用Spark+Cassandra打造高性能数据分析平台（二）

【导读】笔者（
许鹏）看Spark源码的时间不长，记笔记的初衷只是为了不至于日后遗忘。在源码阅读的过程中秉持着一种非常简单的思维模式，就是努力去寻找一条贯穿全局的主线索。在笔者看来，Spark中的线索就是如何让数据的处理在分布式计算环境下是高效，并且可靠的。

在对Spark内部实现有了一定了解之后，当然希望将其应用到实际的工程实践中，这时候会面临许多新的挑战，比如选取哪个作为数据仓库，是HBase、MongoDB还是Cassandra。即便一旦选定之后，在实践过程还会遇到许多意想不到的问题。

要想快速的解决开发及上线过程中遇到的系列问题，还需要具备相当深度的Linux知识，恰巧之前工作中使用Linux的经验在大数据领域中还可以充分使用。

笔者不才，就遇到的一些问题，整理出来与诸君共同分享。下文为本系列文章的第二部分（点击访问本系列文章开篇）：

Cassandra高并发数据读取实现剖析

本文就spark-cassandra-connector的一些实现细节进行探讨，主要集中于如何快速将大量的数据从Cassandra中读取到本地内存或磁盘。

数据分区

存储在Cassandra中的数据一般都会比较多，记录数在千万级别或上亿级别是常见的事。如何将这些表中的内容快速加载到本地内存就是一个非常现实的问题。

解决这一挑战的思路从大的方面来说是比较简单的，那就是将整张表中的内容分成不同的区域，然后分区加载，不同的分区可以在不同的线程或进程中加载，利用并行化来减少整体加载时间。

顺着这一思路出发，要问的问题就是Cassandra中的数据如何才能分成不同的区域。

不同于MySQL，在Cassandra中是不存在Sequence Id这样的类型的，也就是说无法简单的使用seqId来指定查询或加载的数据范围。

既然没有SequenceID，在Cassandra中是否就没有办法了呢？答案显然是否定的，如果只是仅仅支持串行读取，Cassandra早就会被扔进垃圾桶了。

数据分区在Cassandra中至少可以通过两种途径实现 ，一是通过token range，另一个是slice range。这里主要讲解利用token range来实现目的。

1. Token Range

Cassandra将要存储的记录存储在不同的区域中，判断某一记录具体存储在哪个区域的依据是partition key的Hash值。

在Cassandra 1.2之前，组成Cassandra集群的所有节点（Node），都需要手动指定该节点的Hash值范围也就是Token Range。

手工计算Token Range显然是很繁琐，同时也不怎么容易维护，在Cassandra 1.2之后，引进了虚拟节点（vnode）的概念，主要目的是减少不必要的人工指定，同时也将token range的划分变得更为细粒度。比如原先手工指定token range，只能达到10000这样一个精度，而有了vnode之后，默认安装是每一个物理节点上有256个虚拟节点，这样子的话每一个range的范围就是10000/256，这样变的更为精细。

有关token range的信息存储在cassandra的system命名空间(keyspace)下的local和peers两张表中。其中local表示本节点的token range情况，而peers表示集群中其它节点的token range情况。这两张表中的tokens字段就存储有详细的信息。如果集群中只由一台机器组成，那么peers中的就会什么内容都没有。

简单实验，列出本节点的token range：

use system;
desc table local;
select tokens from local;

2. Thrift接口

Token Range告诉我们Cassandra的记录是分片存储的，也就意味着可以分片读取。现在的问题转换成为如何知道每一个Token Range的起止范围。

Cassandra支持的Thrift接口中describe_ring就是用来获取token range的具体起止范围的。我们常用的nodetool工具使用的就是thrift接口，nodetool 中有一个describering指令使用的就是describe_ring原语。

可以做一个简单的实验，利用nodetool来查看某个keyspace的token range具体情况。

nodetool -hcassandra_server_addr describering keyspacename

注意将cassandra_server和keyspacename换成实际的内容。

Spark-Cassandra-Connector

在第一节中讲解了Cassandra中Token Range信息的存储位置，以及可以使用哪些API来获取token range信息。

接下来就分析spark-cassandra-connector是如何以cassandra为数据源将数据加载进内存的。

以简单的查询语句为例，假设用户要从demo这个keyspace的tableX表中加载所有数据，用CQL来表述就是：

select * from demo.tableX

上述的查询使用spark-cassandra-connector来表述就是：

sc.cassandraTable(“demo”,”tableX”)

尽管上述语句没有触发Spark Job的提交，也就是说并不会将数据直正的从Cassandra的tableX表中加载进来，但spark-cassandra-connector还是需要进行一些数据库的操作。要解决的主要问题就是schema相关。

cassandraTable(“demo”,”tableX”)只是说要从tableX中加载数据，并没有告诉connector有哪些字段，每个字段的类型是什么。这些信息对后面使用诸如get[String](“fieldX”)来说却是非常关键的。

为了获取字段类型信息的元数据，需要读取system.schema_columns表，利用如下语句可以得到schema_columns表结构的详细信息：

desc table system.schema_columns

如果在conf/log4j.properties中将日志级别设置为DEBUG，然后再执行sc.cassandraTable语句就可以看到具体的CQL查询语句是什么。
1. CassandraRDDPartitioner

Spark-cassandra-connector添加了一种新的RDD实现，即CassandraRDD。我们知道对于一个Spark RDD来说，非常关键的就是确定getPartitions和compute函数。

getPartitions函数会调用CassandraRDDPartitioner来获取分区数目：

override def getPartitions: Array[Partition] = {
verify // let's fail fast
val tf = TokenFactory.forCassandraPartitioner(cassandraPartitionerClassName)
val partitions = new CassandraRDDPartitioner(connector, tableDef, splitSize)(tf).partitions(where)
logDebug(s"Created total ${partitions.size} partitions for $keyspaceName.$tableName.")
logTrace("Partitions: \n" + partitions.mkString("\n"))
partitions
}

CassandraRDDPartitioner中的partitions的处理逻辑大致如下：

首先确定token range，使用describe_ring
然后根据Cassandra中使用的Partitioner来确定某一个token range中可能的记录条数，这么做的原因就是为进一步控制加载的数据，提高并发度。否则并发度就永远是256了，比如有一个物理节点，其中有256个vnodes，也就是256个token分区。如果每个分区中大致的记录数是20000，而每次加载最大只允许1000的话，整个数据就可以分成256x2=512个分区。
对describeRing返回的token range进一步拆分的话，需要使用splitter，splitter的构建需要根据keyspace中使用了何种Partitioner来决定，Cassandra中默认的Partitioner是Murmur3Partitioner，Murmur3Hash算法可以让Hash值更为均匀的分布到不同节点。
splitter中会利用到配置项spark.cassandra.input.split.size和spark.cassandra.page.row.size，分别表示一个线程最多读取多少记录，另一个表示每次读取多少行。

partitions的源码详见CasssandraRDDParitioner.scala

compute函数就利用确定的token的起止范围来加载内容，这里在理解的时候需要引起注意的就是flatMap是惰性执行的，也就是说只有在真正需要值的时候才会被执行，延迟触发。

数据真正的加载是发生在fetchTokenRange函数，这时使用到的就是Cassandra Java Driver了，平淡无奇。

2. fetchTokenRange

fetcchTokenRange函数使用Cassandra Java Driver提供的API接口来读取数据，利用Java API读取数据一般遵循以下步骤：

val cluster = ClusterBuilder.addContactPoint(“xx.xx.xx.xx”).build
val session = cluster.connect
val stmt = new SimpleStatement(queryCQL)
session.execute(session)
session.close
cluster.close

addContactPoint的参数是cassandra server的ip地址，在后面真正执行cql语句的时候，如果集群有多个节点构成，那么不同的cql就会在不同的节点上执行，自动实现了负载均衡。可以在addContactPoint的参数中设定多个节点的地址，这样可以防止某一节点挂掉，无法获取集群信息的情况发生。

session是线程安全的，在不同的线程使用同一个session是没有问题的，建议针对一个keySpace只使用一个session。

3. RDD中使用Session

在Spark RDD中是无法使用SparkContext的，否则会形成RDD嵌套的现象，因为利用SparkContext很容易构造出RDD，如果在RDD的函数中如map中调用SparkContext创建一个新的RDD，则形成深度嵌套进而导致Spark Job有嵌套。

但在实际的情况下，我们需要根据RDD中的值再去对数据库进行操作，那么有什么办法来打开数据库连接呢？

解决的办法就是直接使用Cassandra Java Driver而不再使用spark-cassandra-connector的高级封装，因为不能像这样子来使用cassandraRDD。

sc.cassandraRDD(“ks”,”tableX”)
.map(x=>sc.cassandraRDD(“ks”,”tableX”).where(filter))

如果是直接使用Cassandra Java Driver，为了避免每个RDD中的iterator都需要打开一个session，那么可以使用foreachPartition函数来进行操作，减少打开的session数。

val  rdd1 = sc.cassandraTable(“keyspace”,”tableX”)
rdd1.foreachPartition( lst => {
val cluster = ClusterBuilder.addContactPoint(“xx.xx.xx.xx”).build
val session = cluster.connect
while ( iter.hasNext ) {
val  elem = iter.next
//do something by using session and elem
}
session.close
cluster.close
})

其实最好的办法是在外面建立一个session，然后在不同的partition中使用同一个session，但这种方法不行的原因是在执行的时候会需要”Task not Serializable”的错误，于是只有在foreachPartition函数内部新建session。

数据备份

尽管Cassandra号称可以做到宕机时间为零，但为了谨慎起见，还是需要对数据进行备份。

Cassandra提供了几种备份的方法

将数据导出成为json格式
利用copy将数据导出为csv格式
直接复制sstable文件

导出成为json或csv格式，当表中的记录非常多的时候，这显然不是一个好的选择。于是就只剩下备份sstable文件了。

问题是将sstable存储到哪里呢？放到HDFS当然没有问题，那有没有可能对放到HDFS上的sstable直接进行读取呢，在没有经过任务修改的情况下，这是不行的。

试想一下，sstable的文件会被拆分为多个块而存储到HDFS中，这样会破坏记录的完整性，HDFS在存储的时候并不知道某一block中包含有完成的记录信息。

为了做到记录信息不会被拆分到多个block中，需要根据sstable的格式自行提取信息，并将其存储到HDFS上。这样存储之后的文件就可以被并行访问。

Cassandra中提供了工具sstablesplit来将大的sstable分割成为小的文件。

DataStax的DSE企业版中提供了和Hadoop及Spark的紧密结合，其一个很大的基础就是先将sstable的内容存储到CFS中，大体的思路与刚才提及的应该差不多。

对sstable存储结构的分析是一个研究的热门，可以参考如下的链接。

https://www.fullcontact.com/blog/cassandra-sstables-offline/

之所以要研究备份策略是想将对数据的分析部分与业务部分相分离开，避免由于后台的数据分析导致Cassandra集群响应变得缓慢而致前台业务不可用，即将OLTP和OLAP的数据源分离开。

通过近乎实时的数据备份，后台OLAP就可以使用Spark来对数据进行分析和处理。

高级查询 Cassandra+Solr

与传统的RDBMS相比，Cassandra所能提供的查询功能实在是弱的可以，如果想到实现非常复杂的查询功能的，需要将Cassandra和Solr进行结合。

DSE企业版提供了该功能，如果想手工搭建的话，可以参考下面的链接：

http://www.slideshare.net/planetcassandra/an-introduction-to-distributed-search-with-cassandra-and-solr

https://github.com/Stratio/stratio-cassandra开源方面的尝试 Cassandra和Lucene的结合

共享SparkContext

SparkContext可以被多个线程使用，这意味着同个Spark Application中的Job可以同时提交到Spark Cluster中，减少了整体的等待时间。

在同一个线程中， Spark只能逐个提交Job，当Job在执行的时候，Driver Application中的提交线程是处于等待状态的。如果Job A没有执行完，Job B就无法提交到集群，就更不要提分配资源真正执行了。

那么如何来减少等待时间呢，比如在读取Cassandra数据的过程中，需要从两个不同的表中读取数据，一种办法就是先读取完成表A与读取表B，总的耗时是两者之和。

如果利用共享SparkContext的技术，在不同的线程中去读取，则耗时只是两者之间的最大值。

在Scala中有多种不同的方式来实现多线程，现仅以Future为例来说明问题：

val ll  = (1 to 3 toList).map(x=>sc.makeRDD(1 to 100000 toList, 3))
val futures = ll.map ( x => Future {
x.count()
})
val fl = Future.sequencce(futures)
Await.result(fl,3600 seconds)

简要说明一下代码逻辑
创建三个不同的RDD
在不同的线程(Future)中通过count函数来提交Job
使用Await来等待Future执行结束

更多《问底》内容

【问底】严澜：数据挖掘入门（一）——分词

【问底】Yao Yu谈Twitter的百TB级Redis缓存实践

【问底】静行：FastJSON实现详解

【问底】李平：大型网站的灵魂——性能

【问底】许鹏：使用Spark+Cassandra打造高性能数据分析平台（一）

【问底】许鹏：使用Spark+Cassandra打造高性能数据分析平台（二）

【问底】徐汉彬：大规模网站架构的缓存机制和几何分形学

【问底】徐汉彬：亿级Web系统搭建——单机到分布式集群