Apache Spark技术实战（四）spark-submit常见问题及其解决 &CassandraRDD高并发数据读取实现剖析

<一>spark-submit常见问题及其解决

概要

编写了独立运行的Spark Application之后，需要将其提交到Spark Cluster中运行，一般会采用spark-submit来进行应用的提交，在使用spark-submit的过程中，有哪些事情需要注意的呢？

本文试就此做一个小小的总结。

spark-defaults.conf

Spark-defaults.conf的作用范围要搞清楚，编辑driver所在机器上的spark-defaults.conf，该文件会影响 到driver所提交运行的application，及专门为该application提供计算资源的executor的启动参数

只需要在driver所在的机器上编辑该文件，不需要在worker或master所运行的机器上编辑该文件

举个实际的例子

spark.executor.extraJavaOptions	   -XX:MaxPermSize=896m
spark.executor.memory		   5g
spark.serializer        org.apache.spark.serializer.KryoSerializer
spark.cores.max		32
spark.shuffle.manager	SORT
spark.driver.memory	2g

 上述配置表示为该application提供计算资源的executor启动时, heap memory需要有5g。

这里需要引起注意的是，如果worker在加入cluster的时候，申明自己所在的机器只有4g内存，那么为上述的application分配executor是，该worker不能提供任何资源，因为4g<5g，无法满足最低的资源需求。

spark-env.sh

spark-env.sh中最主要的是指定ip地址，如果运行的是master，就需要指定SPARK_MASTER_IP，如果准备运行driver或worker就需要指定SPARK_LOCAL_IP，要和本机的IP地址一致，否则启动不了。

配置举例如下：

export SPARK_MASTER_IP=127.0.0.1
export SPARK_LOCAL_IP=127.0.0.1

启动spark集群

第一步启动master

$SPARK_HOME/sbin/start-master.sh

第二步启动worker

$SPARK_HOME/bin/spark-class org.apache.spark.deploy.worker.Worker spark://master:7077

将master替换成MASTER实际运行的ip地址

如果想在一台机器上运行多个worker(主要是用于测试目的),那么在启动第二个及后面的worker时需要指定—webui-port的内容，否则会报端口已经被占用的错误,启动第二个用的是8083，第三个就用8084，依此类推。

$SPARK_HOME/bin/spark-class org.apache.spark.deploy.worker.Worker spark://master:7077
    –webui-port 8083

 这种启动worker的方式只是为了测试是启动方便，正规的方式是用SPARK_HOME/sbin/start-slaves.sh来启动多个worker，由于涉及到ssh的配置，比较麻烦，我这是图简单的办法。

用SPARK_HOME/sbin/start−slave.sh 来启动worker时有一个默认的前提，即在每台机器上$SPARK_HOME必须在同一个目录。

使用相同的用户名和用户组来启动Master和Worker，否则Executor在启动后会报连接无法建立的错误。

我在实际的使用当中，遇到”no route to host”的错误信息，起初还是认为网络没有配置好，后来网络原因排查之后，忽然意识到有可能使用了不同的用户名和用户组，使用相同的用户名/用户组之后，问题消失。

spark-submit

spark集群运行正常之后，接下来的问题就是提交application到集群运行了。

Spark-submit用于Spark application的提交和运行，在使用这个指令的时候最大的困惑就是如何指定应用所需要的依赖包。

首先查看一下spark-submit的帮助文件：

$SPARK_HOME/bin/submit --help

有几个选项可以用来指定所依赖的库，分别为：

• --driver-class-path driver所依赖的包，多个包之间用冒号(:)分割
• --jars   driver和executor都需要的包，多个包之间用逗号(,)分割

为了简单起见，就通过—jars来指定依赖，运行指令如下：

$SPARK_HOME/bin/spark-submit –class 应用程序的类名 \
--master spark://master:7077 \
--jars 依赖的库文件 \
spark应用程序的jar包

需要提醒的时，这些上传到worker的文件，需要定时做手工清理，否则会占用许多磁盘空间

问题1

由于Spark在计算的时候会将中间结果存储到/tmp目录，而目前linux又都支持tmpfs，其实说白了就是将/tmp目录挂载到内存当中。

那么这里就存在一个问题，中间结果过多导致/tmp目录写满而出现如下错误

No Space Left on the device

解决办法就是针对tmp目录不启用tmpfs,修改/etc/fstab

问题2

有时可能会遇到java.lang.OutOfMemory, unable to create new native thread的错误，导致这种错误的原因比较多。

有一种情况并非真的是内存不足引起的，而是由于超出了允许的最大文件句柄数或最大进程数。

排查的步骤就是查看一下允许打开的文件句柄数和最大进程数，如果数值过低，使用ulimit将其调高之后，再试试问题是否已经解决。

ulimit -a

修改允许打开的最大进程数

ulimit -u 65535

修改允许打开的文件句柄

ulimit -n 65535

spark-shell

上面讲述了spark-submit提交Spark Application如何解决依赖库的问题，那如果是spark-shell的话，该怎么办呢？

spark-shell的话，利用--driver-class-path选项来指定所依赖的jar文件，注意的是--driver-class-path后如果需要跟着多个jar文件的话，jar文件之间使用冒号(:)来分割。

小结

本文部分内容已由本人徽沪一郎在CSDN中通过”使用Spark+Cassandra打造高性能数据分析平台“发表。

<二>CassandraRDD高并发数据读取实现剖析

概要

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

数据分区

存储在 Cassandra 中数据的一般都会比较多,记录数在千万级别或上亿级别是常见的事。如何将这些表中的内容快速加载到本地内存就是一个非常现实的问题。
解决这一挑战的思路从大的方面来说是比较简单的,那就是将整张表中的内容分成不同的区域,然后分区加载,不同的分区可以在不同的线程或进程中加载,利用并行化来减少整体加载时间。

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

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

既然没有 SequenceID,在 Cassandra 中是否就没有办法了呢?答案显然是否定的,如果只是仅仅支持串行读取,Cassandra 早就会被扔进垃圾堆里了。
数据分区的办法在 Cassandra 中至少有两种办法可以达到,一是通过 token range,另一个是 slice range。这里主要讲解利用 token range 来实现目的。

token range

Cassandra将要存储的记录存储在不同的区域中，判断某一记录具体存储在哪个区域的依据是主键的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;

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

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查询语句是什么。

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的处理逻辑大致如下：

partitions的源码详见CasssandraRDDParitioner.scala。

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

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

fetchTokenRange

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

val cluster = Cluster.Builder.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.

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提供了几种备份的方法：

1. 将数据导出成为json格式

2. 利用copy将数据导出为csv格式

3. 直接复制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存储结构的分析是一个研究的热门，可以参考如下的链接。

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

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

复杂查询

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

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

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

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

2. 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)

简要说明一下代码逻辑：

1. 创建三个不同的RDD

2. 在不同的线程(Future)中通过count函数来提交Job

3. 使用Await来等待Future执行结束