DolphinDB 与 Spark 的性能对比测试报告

关注
发布于: 2020 年 12 月 17 日
1. 概述﻿
Spark是基于内存计算的通用大数据并行计算框架，内置多种组件，如批处理、流处理、机器学习和图处理。Hive是基于Hadoop的数据仓库，支持类SQL的命令查询，提升了Hadoop的易用性。Spark与Hive、Hadoop通常是搭配使用，利用Hive中的数据分区可以方便地管理和过滤数据，提高查询效率。
﻿
DolphinDB是C++编写的高性能分布式时序数据库，内置高吞吐低延迟的列式内存引擎，集成了功能强大的编程语言，支持类Python和SQL的脚本语言，可以直接在数据库中进行复杂的编程和运算。DolphinDB内部用Data Source来抽象分区数据。在Data Source之上，可以完成SQL，机器学习，批处理和流处理等计算任务。一个Data Source既可以是内置的数据库分区，也可以是外部数据。如果Data Source是内置数据库的分区，大部分计算都可以在本地完成，极大地提升了计算和查询效率。
﻿
本报告将对DolphinDB、Spark直接访问HDFS（Spark+Hadoop，下文称为Spark）、Spark通过Hive组件访问HDFS（Spark+Hive+Hadoop，下文称为Spark+Hive）三者进行性能对比测试。测试内容包括数据导入、磁盘空间占用、数据查询以及多用户并发查询。通过对比测试，我们能更深入的了解影响性能的主要因素，以及不同工具的最佳应用场景。
﻿
2. 环境配置﻿
2.1 硬件配置﻿
本次测试使用了两台配置完全相同的服务器（机器1，机器2），各个配置参数如下：
﻿
主机：DELL PowerEdge R730xd
﻿
CPU：Intel Xeon® CPU E5-2650 v4（24核 48线程 2.20GHz）
﻿
内存：512 GB （32GB × 16, 2666 MHz）
﻿
硬盘：17T HDD （1.7T × 10, 222 MB/s 读取；210 MB/s 写入）
﻿
网络：万兆以太网
﻿
OS： CentOS Linux release 7.6.1810 (Core)
﻿
2.2 集群配置﻿
测试的DolphinDB版本为Linux v0.95。测试集群的控制节点部署在机器1上，每台机器上各部署三个数据节点，共六个数据节点。每个数据节点配置8个worker，7个executor，24G内存。
﻿
测试的Spark版本为2.3.3，搭载Apache Hadoop 2.9.0。Hadoop与Spark配置为完全分布式模式，机器1为Master，并且在机器1、机器2上都具有Slave。Hive的版本是1.2.2，机器1、机器2上都具有Hive。元数据存储在机器1上的MySql数据库中。Spark 与Spark + Hive使用Standalone模式下的client 方式来提交应用。
﻿
测试时，DolphinDB、Spark、Spark+Hive均配置6块硬盘，不同并发数下使用的CPU、内存总和都相同，都是48个线程，144G内存。Spark与Spark+Hive使用的资源只是对于特定的应用，每个应用有6个executor，在多用户并发的情况下，Spark、Spark+Hive单个用户使用的资源会随着用户数量增多而减少。不同并发数下每个用户使用的资源如表1所示。
﻿
表1.Spark、Spark+Hive不同并发数下单用户使用的资源
﻿
﻿
3. 数据集及数据库设计﻿
3.1 数据集﻿
测试数据集是纽约证券交易所（NYSE）提供的TAQ数据集，包含 8000 多支股票在2007.08.01-2007.08.31一个月内的Level 1报价数据，包含交易时间, 股票代码, 买入价, 卖出价, 买入量, 卖出量等报价信息。数据集中共有 65 亿（6,561,693,704）条报价记录，一个 CSV 中保存一个交易日的记录，该月共23个交易日，未压缩的 23个CSV 文件共计 277 GB。
﻿
数据来源：https://www.nyse.com/market-data/historical
﻿
3.2 数据库设计﻿
表2. TAQ在各个系统中的数据类型。
﻿
﻿
在DolphinDB中，我们按照date、symbol列组合分区，第一分区使用日期DATE来进行值分区，共23个分区，第二分区使用股票代码SYMBOL来进行范围分区，分区数量100个，每个分区大约120M左右。
﻿
Spark存储在HDFS上的数据以23个csv对应23个目录。Spark+Hive采用两层分区，第一层分区使用日期DATE列进行静态分区，第二层分区使用股票代码SYMBOL进行动态分区。
﻿
具体脚本见附录。
﻿
4. 数据导入和查询测试﻿
4.1 数据导入测试﻿
原始数据均匀地分布在两台服务器的6个硬盘上，这样可以充分利用集群中的所有资源。DolphinDB通过异步多节点的方式并行导入数据，Spark与Spark+Hive并行启动6个应用来读取数据，把数据存储到HDFS中。各个系统导入数据的时间如表3所示。各个系统中数据占用的磁盘空间如表4所示。数据导入脚本见附录。
﻿
表3. DolphinDB、Spark、Spark+Hive导入数据时间
﻿
﻿
表4. DolphinDB、Spark、Spark+Hive中数据占用的磁盘空间
﻿
﻿
DolphinDB的导入性能明显优于Spark和Spark+Hive，是Spark的4倍左右，是Spark + Hive的6倍左右。DolphinDB使用C++编写并且内部有很多优化，极大地利用了磁盘的IO。
﻿
DolphinDB占用的磁盘空间大于Spark与Spark+Hive，大约是他们的2倍，这是因为Spark和Spark+Hive在Hadoop上都使用Parquet格式，Parquet格式通过Spark写入到Hadoop上默认使用snappy压缩。
﻿
4.2 数据查询测试﻿
为保证测试公平，每个查询语句要进行多次测试，每次测试之前均通过 Linux 系统命令分别清除系统的页面缓存、目录项缓存和硬盘缓存。DolphinDB还清除其内置缓存。
﻿
表5中的查询语句涵盖了大部分查询场景，包含分组、排序、条件、聚合计算、点查询、全表查询，用来评估DolphinDB、Spark、Spark+Hive在不同用户数量提交下的性能。
﻿
表5. DolphinDB、Spark、Spark+Hive查询语句
﻿
﻿
4.2.1 DolphinDB与Spark单用户查询测试﻿
以下是DolphinDB与Spark单用户查询的结果，结果中的耗时为查询8次的平均用时。
﻿
表6. DolphinDB、Spark单用户查询结果
﻿
﻿
从结果可以看出，DolphinDB的查询性能是Spark+HDFS的200倍左右。查询Q1到Q6都是以DolphinDB的分区字段为过滤条件，DolphinDB只需要加载指定分区的数据，无需全表扫描，而Spark从Q1到Q6都需要全表扫描，耗费大量的时间。对于查询Q7，DolphinDB和Spark都需要全表扫描，但是DolphinDB只加载相关的列，无需加载所有列，而Spark则需要加载所有数据。由于Query运行时间被数据加载主导，DolphinDB和Spark的性能差距没有之前的查询语句的大。
﻿
4.2.2 DolphinDB与Spark+Hive单用户查询测试﻿
由于DolphinDB的数据经过分区，且在查询的时候实现谓词下推，效率明显高于Spark。此处我们使用Spark搭载Hive组件来访问HDFS，对比DolphinDB和Spark+Hive的查询性能。以下是DolphinDB、Spark+Hive单用户查询的结果，结果中的耗时为查询8次的平均用时。
﻿
表7. DolphinDB、Spark+Hive单用户查询结果
﻿
﻿
结果显示，DolphinDB的查询性能明显优于Spark+Hive，是Spark+Hive的数十倍。与表6的结果相比，Spark+Hive的查询速度比Spark要快得多，DolphinDB具有的优势明显下降了很多。这是因为Hive对数据进行分区，且在查询语句的条件带有分区字段的时候，只加载部分数据，实现数据过滤，提高效率。查询语句Q7扫描全表的时候会出现内存溢出。
﻿
DolphinDB、Spark+Hive都对数据进行了分区，且在加载数据时都可以实现谓词下推，达到数据过滤的效果，但是DolphinDB的查询速度优于Spark+Hive。这是因为Spark+Hive区读取HDFS上的数据是不同系统之间的访问，数据要经过序列化、网络传输、反序列化的过程，非常耗时，从而影响性能。DolphinDB的大部分计算都在本地完成，减少了数据传输，因此更加高效。
﻿
4.2.3 DolphinDB与Spark计算能力对比﻿
上面DolphinDB分别与Spark、Spark+Hive的查询性能对比，由于数据分区、查询时的数据过滤以及传输影响了Spark的性能，因此这里我们先把数据加载到内存中，再进行相关的计算，比较DolphinDB和Spark+Hive。我们省略了Spark+Hive，因为使用Hive只是为了数据过滤，读取HDFS上的数据更加高效，这里的测试数据已经在内存中。
﻿
表8是测试计算能力的语句。每次测试都包含两个语句，第一个语句是把数据加载到内存中，第二个语句是对内存中的数据进行计算。DolphinDB会自动缓存数据，Spark则通过自己的默认缓存机制重新创建一个临时表TmpTbl。
﻿
表8. DolphinDB与Spark计算能力对比语句
﻿
﻿
以下是DolphinDB与Spark计算能力的测试结果，结果中的耗时是测试5次的平均用时。
﻿
表9. DolphinDB与Spark计算能力测试结果
﻿
﻿
由于数据已经在内存中，对比表6，Spark使用的时间大幅度减少，但是DolphinDB的计算能力仍然比Spark优越。DolphinDB用C++编写，自己管理内存，比起Spark使用JVM来管理内存更加高效。另外，DolphinDB内置了更高效的算法，提高了计算性能。
﻿
DolphinDB的分布式计算以分区为单位，计算指定内存的数据。Spark加载整个HDFS上的块，一个数据块包含了具有不同symbol值的数据，虽然缓存，但是仍然要筛选，所以在Q1与Q2的比值较大。Spark计算时使用的广播变量是经过压缩的，传输到其他的executor上再解压影响性能。
﻿
4.2.4 多用户并发查询﻿
我们使用表5中的查询语句，对DolphinDB、Spark、Spark+Hive进行多用户并发查询测试。以下是测试结果，结果中的耗时是查询8次的平均用时。
﻿
表10. DolphinDB、Spark、Spark+Hive多用户并发查询结果
﻿
﻿
图1. DolphinDB、Spark多用户查询结果对比
﻿
﻿
图2. DolphinDB、Spark+Hive多用户查询结果对比
﻿
﻿
从上面的结果可以看出，随着并发数量的增加，三者的查询时间逐渐增加。当达到8个用户并发的时候Spark性能较之前少量的用户并发情况下显著下降，Spark 在执行Q7的时候会导致worker死亡。Spark+ Hive在多用户访问的时候与DolphinDB一样，基本保持稳定，但是执行Q7查询语句的一直会出现内存溢出的异常。
﻿
Spark+ Hive的查询配置与Spark 一样，因为有分区的作用，并且可以过滤数据，查询数据量比较小，所以效率相对于Spark扫描全部数据比较好。
﻿
DolphinDB在并发查询中性能明显优于Spark 与Spark+ Hive，从上图可以看出在多用户并发访问情况下，随着用户数量的增加，DolphinDB相对于Spark 的优势几乎是线性增长，相对于Spark + Hive 的优势基本保持不变，体现了有数据分区在查询的时候实现数据过滤的重要性。
﻿
DolphinDB在多用户并发的情况下实现了多用户的数据共享，不像Spark 的数据只是针对于具体的应用。所以在8个并发用户的情况下，Spark 每个用户分配到的资源比较少，性能显著下降。DolphinDB的数据共享可以减少资源的使用，在有限的资源下，把更多的资源留给用户计算使用，提高用户并发的效率，增大用户的并发数量。
﻿
5. 小结﻿
在数据的导入方面，DolphinDB可以并行加载，Spark与Spark+Hive 则通过多个应用同时加载来导入数据。DolphinDB的导入速度是Spark 和Spark+ Hive 的4-6倍。在磁盘空间上，DolphinDB占用的磁盘空间是Spark与Spark+ Hive在Hadoop上占用的磁盘空间的两倍左右，Spark与Spark + Hive使用了snappy压缩。
﻿
在数据的SQL查询方面，DolphinDB的优势更加明显。优势主要来自四个方面：（1）本地化计算，（2）分区过滤，（3）优化的内存计算，（4）跨会话的数据共享。在单用户查询情况下，DolphinDB的查询速度是Spark的几倍到上百倍，是Spark+ Hive 的几十倍。Spark 读取HDFS 是不同的系统之间的调用，其中包含了数据的序列化，网络，反序列化非常消耗时间，且占据很多的资源。DolphinDB的SQL查询大部分是本地化计算，大幅减少了数据传输和加载的时间。Spark+ Hive 相对与Spark速度提升很大，主要是因为Spark + Hive只扫描相关分区的数据，实现了数据的过滤。在剔除本地化和分区过滤的因素后（即所有数据已经在内存中），DolphinDB的计算能力仍然优于Spark数倍。DolphinDB基于分区的分布式计算效率很高，且对内存的管理比Spark基于JVM的管理更加优秀。Spark的多用户并发会随着用户数量的增多效率逐渐下降，在查询大数据量的时候用户过多导致worker 死亡。Spark + Hive的多用户并发相对比较稳定，但是加载数据过大会出现内存溢出错误。 多用户情况下, DolphinDB可以实现数据的共享，从而减少加载数据使用的资源，查询速度是Spark的数百倍，是Spark+Hive 的几十倍。随着用户数量的增加，DolphinDB相对于Spark的性能优势更加明显。涉及到分区查询的情况下，Spark+ Hive与DolphinDB显著提高查询性能。
﻿
Spark是一个非常优秀的通用分布式计算引擎，在SQL查询、批处理、流处理、机器学习等方面均有上佳表现。但由于SQL查询通常只需要对数据计算一次，相对于机器学习需要上百次的迭代，内存计算的优势无法充分体现。因此，我们更建议将Spark用于计算密集型的机器学习。
﻿
在测试过程中，我们也发现DolphinDB是一个非常轻量级的实现，集群的搭建简单快速， Spark + Hive+ Hadoop 集群安装配置非常复杂。
﻿
附录﻿
附录1. 数据预览﻿
﻿
附录2. Hive创建表语句﻿
CREATE TABLE IF NOT EXISTS TAQ (time TIMESTAMP, bid DOUBLE, ofr DOUBLE, bidsiz INT, ofrsiz INT, mode INT, ex TINYINT, mmid STRING）PARTITIONED BY (date DATE, symbol STRING) STORED AS PARQUET;
1
﻿
附录3.﻿
DolphinDB导入数据脚本：
﻿
fps1、fps2分别代表机器1、2上所有的csv路径的vector
fps是包含fps1和fps2 的vector
allSites1、allSites2 分别代表机器1、2上数据节点名称的vector
allSite 是包含 allSites1和allSites2的vector
DATE_RANGE=2007.07.01..2007.09.01
date_schema=database('', VALUE, DATE_RANGE)
symbol_schema=database('', RANGE, buckets)
db=database(FP_DB, COMPO,[date_schema,symbol_schema])
taq = db.createPartitionedTable(schema, `taq, `date`symbol)
for(i in 0..1){
	for(j in 0..(size(fps[i])-1))  {
		rpc(allSites[i][j] % size(allSite[i])],submitJob,"loadData" , "loadData" ,loadTextEx{database(FP_DB), "taq", `date`symbol, fps[i][j]} )
	}
}
1234567891011121314
﻿
Spark与Hive导入数据的配置：
﻿
--master local[8]
--executor-memory 24G
﻿