巨杉Tech | SparkSQL+SequoiaDB 性能调优策略

当今时代，企业数据越发膨胀。数据是企业的价值，但数据处理也是一种技术挑战。在海量数据处理的场景，即使单机计算能力再强，也无法满足日益增长的数据处理需求。所以，分布式才是解决该类问题的根本解决方案。而在分布式领域，有两类典型产品，分别是分布式存储和分布式计算。用户只有将两者的特性充分利用，才可以真正发挥分布式架构的存储和计算能力。

本文介绍 SequoiaDB（分布式存储）和 Spark（分布式计算）两款产品的对接使用，以及在海量数据场景下如何提高统计分析性能。

 

01  SequoiaDB 与 SparkSQL 介绍

SequoiaDB 是一款开源的金融级分布式关系型数据库，支持标准 SQL 和事务功能，支持复杂索引查询、与 Hadoop、Hive、Spark 都有较深度的集成。SequoiaDB 在分布式存储功能上，较一般的大数据产品能提供更多的数据切分规则，包括：水平切分、范围切分、主子表切分（类似 partition 分区）和多维切分方式，用户可以根据不用的场景选择相应的切分方式，以提高系统的存储能力和操作性能。

Spark 近年来发展特别迅猛，使用 SparkSQL 做大数据处理和分析的开发者越来越多。SparkSQL 是 Spark 产品中一个组成部分，SQL 的执行引擎使用 Spark 的 RDD 和 Dataframe 实现。

SparkSQL 和另外一款流行的大数据 SQL 产品— Hive 有相似之处，但是两款产品还是有本质上的区别，最大的不同点在于执行引擎，Hive 默认支持 Hadoop 和 Tez 计算框架，而 SparkSQL 只支持 Spark RDD 计算框架，但是 SparkSQL 的拥有更加深度的执行计划优化和处理引擎优化。

 

02 SequoiaDB 与 SparkSQL 如何整合？

Spark 本身是一款分布式计算框架。它不像 Hadoop 一样，同时为开发者提供分布式计算和分布式存储，而是开放了存储层的开发接口，只要开发者按照 Spark 的接口规范实现了接口方法，任何存储产品都可以成为 Spark 数据计算的来源，同时也包括 SparkSQL 的数据来源。

SequoiaDB 是一款分布式数据库，能够为用户存储海量的数据，但是如果要对海量数据做统计、分析，还是需要借助分布式计算框架的并发计算性能，提高计算效率。所以 SequoiaDB 为 Spark开发了 SequoiaDB for Spark 的连接器，让 Spark 支持从SequoiaDB 中并发获取数据，再完成相应的数据计算。

Spark 和 SequoiaDB 对接方式比较简单，用户只要将 SequoiaDB for Spark 连接器 spark-sequoiadb.jar 和 SequoiaDB 的 java 驱动 sequoiadb.jar 加入到每个 Spark Worker 的 CLASSPATH 中即可。

例如，用户希望 SparkSQL 对接到 SequoiaDB，可以为 spark-env.sh 配置文件中增加 SPARK_CLASSPATH 参数，如果该参数已经存在，则将新 jar 包添加到 SPARK_CLASSPATH 参数上，如：

SPARK_CLASSPATH="/media/psf/mnt/sequoiadb-driver-2.9.0-SNAPSHOT.jar:/media/psf/mnt/spark-sequoiadb_2.11-2.9.0-SNAPSHOT.jar"

用户修改完 spark-env.sh 配置后，重启 spark-sql 或者 thriftserver 就完成了 Spark 和 SequoiaDB 的对接。

 

03 SequoiaDB 与 SparkSQL 性能优化

Spark SQL+SequoiaDB 的性能优化将会从 connector 计算技术原理、SparkSQL 优化、SequoiaDB 优化和 connector 参数优化4个方面进行介绍。

 

1 SequoiaDB for SparkSQL

A） connector 工作原理

Spark 产品虽然为用户提供了多种功能模块，但是都只是数据计算的功能模块。Spark 产品本身没有任何的存储功能，在默认情况下，Spark 是从本地文件服务器或者 HDFS 上读取数据。而 Spark 也将它与存储层的接口开放给广大开发者，开发者只要按照 Spark 接口规范实现其存储层连接器，任何数据源均可称为 Spark 计算的数据来源。

下图为 Spark worker 与存储层中 datanode 的关系。

图1

Spark 计算框架与存储层的关系，可以从下图中了解其原理。Spark master 在接收到一个计算任务后，首先会与存储层做一次通讯，从存储层的访问快照或者是存储规划中，得到本次计算任务所设计的所有数据的存储情况。存储层返回给 Spark master 的结果为数据存储的 partition 队列。

然后 Spark master 会将数据存储的 partition 队列中的 partition 逐个分配给给 Spark worker。Spark work 在接收到数据的 partition 信息后，就能够了解如何获取计算数据。然后 Spark work 会主动与存储层的 node 节点进行连接，获取数据，再结合 Spark master 下发给 Spark worker 的计算任务，开始数据计算工作。

图2

 

SequoiaDB for Spark 的连接器的实现原理和上述描述基本一致，只是在生成数据计算的 partition 任务时，连接器会根据 Spark 下压的查询条件到 SequoiaDB 中生成查询计划。

如果 SequoiaDB 能够根据查询条件做索引扫描，连接器生成的 partition 任务将是让Spark work 直接连接 SequoiaDB 的数据节点。

如果 SequoiaDB 无法根据查询条件做索引扫描，连接器将获取相关数据表的所有数据块信息，然后根据 partitionblocknum 和 partitionmaxnum 参数生成包含若干个数据块连接信息的 partititon 计算任务。

 

B）  Connector 参数

SequoiaDB for Spark 连接器在 SequoiaDB 2.10之后进行了重构，提高 Spark 并发从SequoiaDB 获取数据的性能，参数也有相应的调整。

用户在 SparkSQL 上创建数据源为 SequoiaDB 的 table，建表模版如下：

create [temporary] [(schema)] using com.sequoiadb.spark options ();

SparkSQL 创表命令的关键字介绍：

temporary 关键字，代表该表或者视图是否为邻时创建的，如果用户标记了temporary 关键字，则该表或者视图在客户端重启后将自动被删除；

建表时用户可以选择不指定表结构，因为如果用户不显式指定表结构，SparkSQL 将在建表时自动检测已经存在数据的表结构；

com.sequoiadb.spark 关键字为 SequoiaDB for Spark connector  的入口类；

options 为 SequoiaDB for Spark connector 的配置参数；

 

SparkSQL 建表例子如下：

create table tableName (name string, id int) using com.sequoiadb.spark options (host "sdb1:11810,sdb2:11810,sdb3:11810", collectionspace "foo", collection "bar", username "sdbadmin", password "sdbadmin");

 

SparkSQL for SequoiaDB 的建表 options 参数列表如下：

表1

 
3.2 SparkSQL 优化

用户如果要使用 SparkSQL 对海量数据做统计分析操作，那么应该从3个方面进行性能调优：

调大 Spark Worker 最大可用内存大小，防止在计算过程中数据超出内存范围，需要将部分数据写入到临时文件上；

增加 Spark Worker 数目，并且设置每个 Worker 均可以使用当前服务器左右 CPU 资源，以提高并发能力；

调整 Spark 的运行参数；

用户可以对 spark-env.sh 配置文件进行设置，SPARK_WORKER_MEMORY 为控制 Worker 可用内存的参数，SPARK_WORKER_INSTANCES 为每台服务器启动多少个 Worker 的参数。

 

如果用户需要调整 Spark 的运行参数，则应该修改 spark-defaults.conf 配置文件，对优化海量数据统计计算有较明显提升的参数有：

spark.storage.memoryFraction, 该参数控制 Worker 多少内存比例用户存储临时计算数据，默认为0.6，代表60%的含义；

spark.shuffle.memoryFraction, 该参数控制计算过程中 shuffle 时能够占用每个 Worker 的内存比例，默认为0.2，代表20%的含义，如果临时存储的计算数据较少，而计算中有较多的 group by, sort, join 等操作，应该考虑将spark.shuffle.memoryFraction 调大，spark.storage.memoryFraction 调小，避免超出内存部分需要写入临时文件中；

spark.serializer, 该参数设置 Spark 在运行时使用哪种序列化方法，默认为 org.apache.spark.serializer.JavaSerializer, 但是为了提升性能，应该选择 org.apache.spark.serializer.KryoSerializer 序列化。

 

3 SequoiaDB 优化

SparkSQL+SequoiaDB 这种组合，由于数据读取是从 SequoiaDB 中进行，所以在性能优化应该考虑三点

尽可能将大表的数据分布式存储，所以建议符合二维切分条件的 table 应该采用主子表+ Hash 切分两种数据均衡方式进行数据分布式存储；

数据导入时，应该避免同时对相同集合空间的多个集合做数据导入，因为同一个集合空间下的多个集合是共用相同一个数据文件，如果同时向相同集合空间的多个集合做数据导入，会导致每个集合下的数据块存储过于离散，从而导致在 Spark SQL 从SequoiaDB 获取海量数据时，需要读取的数据块过多；

如果 SparkSQL 的查询命令中包含查询条件，应该对应地在 SequoiaDB 中建立对应字段的索引。

 
3.4 connector 优化

SequoiaDB for Spark 连接器的参数优化，主要分两个场景，一是数据读，另外一个是数据写入。

数据写入的优化空间较少，只有一个参数可以调整，即bulksize参数，该参数默认值为500，代表连接器向 SequoiaDB 写入数据时，以500条记录组成一个网络包，再向 SequoiaDB 发送写入请求，通常设置 bulksize 参数，以一个网络包不超过2MB为准。

数据读取的参数优化，用户则需要关注 partitionmode, partitionblocknum 和 partitionmaxnum 三个参数。

    partitionmode，连接器的分区模式，可选值有single、sharding、datablock、auto，默认值为auto，代表连接器智能识别。

    1. single 值代表 SparkSQL 在访问 SequoiaDB 数据时，不考虑并发性能，只用一个线程连接 SequoiaDB 的 Coord 节点，一般该参数在建表做表结构数据抽样时采用；

    2. sharding 值代表 SparkSQL 访问 SequoiaDB 数据时，采用直接连接 SequoiaDB 各个 datanode 的方式，该参数一般采用在 SQL 命令包含查询条件，并且该查询可以在 SequoiaDB 中使用索引查询的场景；

    3. datablock 值代表 SparkSQL 访问 SequoiaDB 数据时，采用并发连接 SequoiaDB 的数据块进行数据读取，该参数一般使用在SQL命令无法在 SequoiaDB 中使用索引查询，并且查询的数据量较大的场景；

    4. auto 值代表 SparkSQL 在向 SequoiaDB 查询数据时，访问 SequoiaDB 的方式将由连接器根据不同的情况分析决定。

    partitionblocknum，该参数只有在 partitionmode=datablock 时才会生效，代表每个 Worker 在做数据计算时，一次获取多少个 SequoiaDB 数据块读取任务，该参数默认值为4。如果 SequoiaDB 中存储的数据量较大，计算时涉及到的数据块较多，用户应该调大该参数，使得 SparkSQL 的计算任务保持在一个合理范围，提高数据读取效率。

    partitionmaxnum，该参数只有在 partitionmode=datablock 时才会生效，代表连接器最多能够生成多少个数据块读取任务，该参数的默认值为1000。该参数主要是为了避免由于 SequoiaDB 中的数据量过大，使得总的数据块数量太大，导致 SparkSQL 的计算任务过多，最后使得总体计算性能下降。

     

04 总结

文章从 Spark、SequoiaDB 以及 SequoiaDB for Spark connector 三个方面向读者们介绍了海量数据下使用 SparkSQL+SequoiaDB 的性能调优方法。

文章中介绍的方法具有一定的参考意义，但是性能调优一直都是最考验技术人员的工作。技术人员在对分布式环境做性能调优时，需要综合考虑多个方面的数据，例如：服务器的硬件资源使用情况、Spark 运行状况、SequoiaDB 数据分布是否合理、连机器的参数设置是否正确、SQL 命令是否有调优的空间等，要想性能提升，重点是要求技术人员找到整个系统中的性能短板，然后通过调整不同的参数或者修改存储方案，从而让系统运行得更加高效。