基于Hadoop和Spark的证券交易数据剖析-上海证券交

日期:2018-03-05 / 人气: / 来源:网络

基于Hadoop和Spark的证券交易数据分析

上海证券交易所

黄寅飞

摘要:

本文对Hive、Impala、Spark SQL三个开源数据仓库产品进行性能测试,对分区技术和列存算法进行比较,并对Spark SQL进行参数配置调优。结果表明Sql on Hadoop开源产品在加载性能、查询性能等方面已达到企业级数据仓库使用要求。本文还结合iPython Notebook对证券数据进行探索式分析和可视化展现。

1 技术概述

证券交易数据属于典型的结构化数据,交易记录存放简短的数字和文本信息,记录长度在数百字节。记录条数较多,按照每天5000万条记录估算,10年的记录总量可超过1000亿条,存储量可达到100TB。采用Sql on Hadoop[1]技术,既可用廉价PC服务器获得良好的容量线性扩展能力,又可提供便于统计分析的SQL接口方便数据应用开发。

1.1 产品概述

Hadoop由Yahoo公司在2006年发布,目标是利用廉价服务器集群搭建面向海量数据的开源计算引擎,同时保证容忍硬件故障的数据高可用性和计算高可靠性,编程语言为Java。Hadoop框架包含Common公共库、HDFS文件系统、YARN调度引擎、MapReduce计算引擎,Hadoop生态圈包含Hive、HBase、Spark、ZooKeeper、Storm等大量开源软件。目前已有Cloudera、Hortonworks、MapR等多家大型企业围绕Hadoop发行版开展经营服务。

Sql on Hadoop的三个代表性开源产品是Hive、Impala[2]和Spark SQL[3],共同特点是数据存储在Hadoop的HDFS分布式文件系统中,通过Hive管理元数据。产品属于OLAP范畴,不建索引,读操作远多于写操作,不支持事务,写操作采用批量执行方式。

Hive是Hadoop生态圈中最早的数据仓库产品,由Facebook在2008年贡献给Apache基金会,编程语言为Java。Hive将SQL语句转换为MapReduce作业提交到Hadoop集群进行分布式计算,再返回执行结果。受限于Hadoop的作业调度开销,Hive性能较差。

Impala是Cloudera公司在2012年发布的MPP(Massively Parallel Processing)数据库产品,目标是为Hadoop提供原生SQL查询引擎,编程语言为C++。Impala借鉴Google的Dremel产品设计思想,使用多层查询树结构,将SQL查询层层分解到下层节点并行执行,在返回结果时各层节点可进行局部聚合。Impala不用MapReduce任务机制,改用MPP机制对数据进行处理,性能较优。

Spark由加州大学伯克利分校AMP实验室(UC Berkeley AMPLab)在2013年贡献给Apache基金会,编程语言为Scala(基于JVM)。Spark SQL是Spark生态圈中的SQL数据引擎,前身为Shark,2014年替代为Spark SQL。借助Spark的RDD内存计算模型和丰富算子,Spark SQL提供对海量数据的SQL查询接口。Spark SQL社区活跃,新版本性能有较大提升。

1.2 技术概述

在Sql on Hadoop产品中应用的关键技术包括分区和列存压缩[4]。分区技术根据特定字段对数据进行划分,缩小扫描范围,辅助查询。列存压缩技术按照列存储并压缩数据,大幅减少IO量。分区参数和压缩算法的合理选择,可有效提升整体性能。

Hive、Impala和Spark SQL依赖HDFS文件系统存储数据文件。HDFS支持行存格式TextFile(文本行)、SequenceFile(键值数组),也支持列存格式RCFile、ORC、Parquet。列存技术将数据表垂直划分,同一列的所有数据连续存储在一起。如SQL查询只涉及个别列,则只需读取对应列内容,可有效减少IO量。因数据按列存储,所以可以针对每列数据的类型采取具有针对性的数据压缩算法,令整体压缩效率大幅提升。

RCFile是Hive推出的列存格式,ORC是针对RCFile提出的优化文件格式。ORC文件包含若干数据带,每个数据带包含若干列记录,以及每列的压缩算法和统计信息。通过数据带中记录信息,可以在查找时跳过不满足条件的列,提高执行效率。

Parquet是Twitter和Cloudera合作推出的列存格式。Parquet文件包含若干行组,每个行组包含若干列块,每个列块包含若干数据页,每个数据页内连续存放列数据,文件尾存储格式信息。Parquet格式对于宽表的查询性能提升明显。

本文对上述产品开展一系列性能试验,评估数据加载性能、单表查询性能、多表查询性能,并对参数配置进行调优。此外,对如何进行探索式分析做了介绍。

2 性能试验

2.1环境参数

试验环境为6台PC服务器(DL380,32核,128GB内存,1TB本地硬盘)构成的千兆网络集群。操作系统为64位RHEL 6.3,安装CDH 5.4.5(Hadoop 2.6.0),包含Hive 1.1.0,Impala 2.2.0,单独安装Spark 1.5.2,共用同一套HDFS文件系统。Spark采用StandAlone集群模式。HDFS副本数配置为2。

数据源为2015年6月至8月间上海证券市场的脱敏交易日志,共65天,30.6亿条订单记录,38.1亿条成交记录,其中订单记录长635字节,成交记录长269字节,总共2.70 TB数据量。Impala内存设为48GB,Spark内存设为48GB。

2.2 加载与查询

交易日志上传到服务器本地磁盘后,经历两步数据加载。第一步是文件加载,通过 hdfs dfs -put 命令,将日志文件加载到HDFS文件系统中,在这一过程中,数据被分块后通过网络进行分布式存储。第二步是表加载,将日志文件映射为外部表,通过Hive的insert overwrite table指令,将交易记录转换为关系表,以Parquet列式压缩格式和日期字段分区存入HDFS文件系统。

加载测试使用65天的成交日志文件,数据总量为955GB,38.1亿条记录。记录不同用例的加载时间和压缩比,见表1。

表1 文件加载与表加载性能

编号

LF1

LT1

LT2

LT3

LT4

LT5

用例

文件加载

表加载TextFile

表加载RcFile

表加载ORC

表加载Parquet

日期分区

表加载Parquet

不分区

耗时

3624秒

4448秒

3264秒

3239秒

5155秒

4819秒

压缩比

n/a

48.7%

43.3%

8.99%

15.18%

15.28%

单个文件(9.9GB)加载时长1分钟44秒,加载速度97.4 MB/s。将分布在6台服务器硬盘上的文件进行并行加载,总耗时1小时24秒,总加载速度269MB/s。并行加载技术可将文件加载速度提升到2.76倍。

表加载阶段, 进一步按照Hive元数据进行ETL转换,数据按列进行压缩,按指定字段进行分区存储。TextFile和RcFile格式简单地滤掉空白字符仅有45%左右的压缩比,ORC和Parquet进行列存压缩可达到8.99%的压缩比,即955GB成交日志压缩后只占用85.9GB的存储空间。

在分区配置方面,按日期分为65个分区。如分区太多会对内存产生较高要求,需小心配置以避免出现OOM(Out Of Memory)错误。

数据导入HDFS并在Hive建立元数据后,使用不同产品进行查询,见表2。单表查询用例包括:个数统计、唯一性统计、排序TOP N、分类汇总、随机查询。多表查询用例包括:维度统计。以Parquet列存压缩、 交易日期分区作为基准。

表2 Sql on Hadoop产品查询性能比较

编号

用例

Hive

Impala

Spark SQL

QB1

个数统计

92.207秒

13.93秒

37.17秒

QB2

唯一性统计

119.343秒

45.00秒

26.832秒

QB3

排序TOP N

n/a

614.19秒

201.02秒

QB4

分类汇总

573.282秒

58.72秒

66.433秒

QB5

随机查询

84.231秒

2.80秒

2.307秒

QB6

维度统计

314.982秒

39.33秒

44.895秒

从表中可见,三个产品中,Hive性能较差(其中QB3用例跑不出来),Impala和Spark SQL性能相当。考虑到Spark SQL在数据格式方面兼容性更好,后续主要基于Spark SQL进行不同参数配置下的性能试验。

2.3 数据格式与分区试验

首先进行数据格式试验,分析列存压缩算法对于Spark SQL查询性能的影响。以交易日期分区作为基准。从表3可以看到,Rcfile列存相对Textfile行存,访问列相比访问整行IO量减少到1/20,查询效率可提高。ORC和Parquet列存压缩将数据量进一步减少到十分之一,查询效率可进一步提高。

表3 不同数据格式下查询性能比较

编号

用例

Textfile

Rcfile

ORC

Parquet

QC1

个数统计

271.534秒

115.29秒

46.528秒

37.17秒

QC2

唯一性统计

185.483秒

134.317秒

40.513秒

26.832秒

QC3

排序TOP N

541.186秒

511.268秒

657.166秒

201.02秒

QC4

分类汇总

289.402秒

140.898秒

157.899秒

66.433秒

QC5

随机查询

2.296秒

2.824秒

1.877秒

2.307秒

QC6

维度统计

227.818秒

175.412秒

138.633秒

44.895秒

其次试验不同分区策略下Impala和Spark SQL的性能,以Parquet作为基准。从表4可以看到,数据分区可对查询性能带来相当大的提升。

表4 不同分区策略下查询性能比较

编号

用例

Impala

不分区

Impala

日期分区

Sparksql

不分区

Sparksql

日期分区

QP1

个数统计

13.46秒

13.93秒

46.715秒

37.17秒

QP2

唯一性统计

45.04秒

45.00秒

30.675秒

26.832秒

QP3

排序TOP N

629.62秒

614.19秒

346.104秒

201.02秒

QP4

分类汇总

60.22秒

58.72秒

117.112秒

66.433秒

QP5

随机查询

3.11秒

2.80秒

2.371秒

2.307秒

QP6

维度统计

39.36秒

39.33秒

94.173秒

44.895秒

2.4 参数配置试验

Spark配置参数较多,对性能影响较大的有内存参数、实例数参数和总核数参数[5]。对不同参数配置下Spark SQL性能进行试验,寻找最优的配置策略。

首先调整SPARK_WORKER_MEMORY参数,将实例数固定为2,总核数固定为32。当实例内存设为6G时,QM3用例会因OOM错误而中止。从表5可以看到,实例内存从9G到12G时,性能提升较大,内存再增加,对于单个任务的执行效率影响不大。

表5 Spark内存参数配置试验

编号

用例

9G

12G

18G

QM1

个数统计

36.258秒

37.17秒

35.086秒

QM2

唯一性统计

25.563秒

26.832秒

26.699秒

QM3

排序TOP N

408.309秒

201.02秒

196.43秒

QM4

分类汇总

66.807秒

66.433秒

65.215秒

QM5

随机查询

3.2秒

2.307秒

2.751秒

QM6

维度统计

43.504秒

44.895秒

44.362秒

实例内存固定为12G,总核数固定为32,调整实例数参数SPARK_WORKER_INSTANCES。从表6可以看到,实例数从1增至2时,性能有一定提升,再增加实例数,对于单个任务性能影响不大。

表6 Spark实例数参数配置试验

编号

用例

1实例

2实例

3实例

QI1

个数统计

33.921秒

37.17秒

36.631秒

QI2

唯一性统计

27.0秒

26.832秒

29.097秒

QI3

排序TOP N

246.428秒

201.02秒

198.137秒

QI4

分类汇总

85.213秒

66.433秒

73.671秒

QI5

随机查询

2.523秒

2.307秒

2.797秒

QI6

维度统计

45.811秒

44.895秒

51.382秒

将实例内存固定为12G,实例数固定为2,调整总核数参数SPARK_WORKER_CORES。从表7可以看到,从4核到16核,性能有显著提升,核数再增加,对单个任务的执行速度影响不大。

表7 Spark总核数参数配置试验

编号

用例

4核

16核

32核

QO1

个数统计

43.106秒

33.683秒

37.17秒

QO2

唯一性统计

31.791秒

30.402秒

26.832秒

QO3

排序TOP N

352.349秒

194.08秒

201.02秒

QO4

分类汇总

122.85秒

71.682秒

66.433秒

QO5

随机查询

2.11秒

2.197秒

2.307秒

QO6

维度统计

74.853秒

44.34秒

44.895秒

2.5 试验小结

经过一系列性能试验,倾向于选择Impala和Spark SQL作为Hadoop系统上用SQL分析结构化数据的工具。数据格式方面Parquet相对更优,数据分区比不分区更优。对Spark SQL,需合理配置内存参数、实例数和总核数参数,以获得优化性能。

3 探索式分析

Impala和Spark SQL可提供基于Hadoop分布式数据上的秒级查询回应。从更多数据源获取产品、席位、会员、营业部信息等基础数据,结合证券交易数据可开展探索式分析,分析数据特征,展现可视化图像,进行数据挖掘。采用PySpark+iPython+MatplotLib技术[6]的组合,可通过网页远程交互并可视化展示。

上证股票共1076支,可按照行业、地域、流通市值等划分板块。从图1中可看到不同板块的成交额饼图,从行业看,制造业、金融业、交通运输业股票交易最为活跃,从地域看,北京、上海、广东省上市股票交易最为活跃,从流通市值看,50-100亿、100-150亿、150-200亿流通市值股票交易最为活跃。相关脚本执行时间分别为64、58、256秒。

基于Hadoop和Spark的证券交易数据分析-上海证券交易所黄寅飞

 

基于Hadoop和Spark的证券交易数据分析-上海证券交易所黄寅飞

 

基于Hadoop和Spark的证券交易数据分析-上海证券交易所黄寅飞

图1 产品交易额分布:行业板块、地域、流通市值

从逐笔成交中可以还原出行情曲线。在逐笔成交数据基础上,结合Spark RDD开发,可生成秒行情,向市场提供历史行情的按需回放功能,如图2所示。相关脚本执行时间为6秒。

基于Hadoop和Spark的证券交易数据分析-上海证券交易所黄寅飞

图2 某股票6月5日逐秒行情

在历史数据中可以根据用户账户查询全部历史行为,绘制用户画像,选取某用户,统计该用户买卖股票的分布,计算用户的买入卖出变化曲线,如图3所示。相关脚本执行时间为75、38秒。

基于Hadoop和Spark的证券交易数据分析-上海证券交易所黄寅飞

 

基于Hadoop和Spark的证券交易数据分析-上海证券交易所黄寅飞

图3 某账户用户画像

在用户交易行为历史数据基础上,可使用Spark MLLib机器学习库[7]对用户进行分类,根据不同类别用户的行为特征,实施差异化的监管风控策略。以股票板块成交金额特征为例,每个账户对应16条特征,5250万活跃账户共计生成特征向量4.7GB,用K均值算法聚类,设K值为5。特征抽取和聚类算法执行时间分别为299、42秒。

4 小结

本文根据证券行业数据特点,提出基于Sql on Hadoop的行业大数据解决方案。通过一系列性能试验,表明Impala、Spark SQL等开源技术在加载性能、查询性能等方面已达到企业级数据仓库使用要求,同时开源社区提供的可视化和机器学习工具可满足数据科学家的探索性分析需要。

致谢

感谢信息中心罗列和樊小泊的数据服务支持。感谢上市公司监管一部刘海波提供的上市股票分类数据。感谢市场监察部李旭和陈雷提供的业务信息。

参考文献

[1] Michael Frampton. Big Data Made Easy: A Working Guide to the Complete Hadoop Toolset. Apress Berkely, CA, USA. 2015

[2] 贾传青. 开源大数据分析引擎:Impala实战. 清华大学出版社. 2015

[3] 王家林. 大数据Spark企业级实战. 电子工业出版社. 2015

[4] 张俊林. 大数据日知录:架构与算法. 电子工业出版社. 2014

[5] H. Karau, A. Konwinski, P. Wendell and M. Zaharia. Learning Spark: Lightning-Fast Big Data Analysis. O’Reilly Media, Inc. 2015

[6] Wes McKinney著,唐学韬等译. 利用Python进行数据分析. 机械工业出版社. 2014

[7] Nick Pentreath著,蔡立宇等译. Spark机器学习. 人民邮电出版社. 2015

大数据分布式系统架构原理概述

大数据分布式系统架构中CAP原理指的是,一致性(Consistency)、可用性(Availability)、分区容忍性(Partition tolerance)这三个要素最多只能同时实现两点,不可能三者兼顾。

分布式系统

作者:管理员




现在致电4006-2991-90 OR 查看更多联系方式 →

Go To Top 回顶部