随着大数据分析挖掘技术不断发展, 数据所蕴含的价值被不断发掘, 数据已成为各行各业社会团体最重要的资源之一.如何高效存储这种来自不同系统, 具有不同格式的多源异构数据, 怎样将这些科研管理数据整合利用, 打破信息孤岛, 进行数据挖掘, 辅助决策分析, 实现数据互通, 支持在线流处理、离线批处理及OLAP、OLTP不同场景的数据分析处理需求, 优化查询效率, 提供经济高效分析计算平台是当下研究热点.

在数字化高速发展的信息时代, 科研管理过程的实质是信息化管理的过程, 是对科研信息资源进行收集、整理、统计、分析并加以利用的过程^[1,2]. 科研管理信息系统广泛应用于高校和科研院所, 已积累形式多样、相互隔离、分布广泛, 标准各异的海量数据, 由于缺乏全局管理、治理维护、统一平台, 无法对科研成果科学评估及辅助决策制定提供有效支持.

本文以中国科学院科研管理态势感知和竞争力分析为研究背景, 依托中国科学院科研与教育态势感知服务项目, 以汇聚全院科研与教育投入、产出、成果、发展等结构化、半结构化和非结构化数据, 构建可扩展高可用大数据仓库、高效OLAP查询分析实际需求为切入点, 聚焦科研管理大数据汇集、存储、分析的需求, 研究、设计和构建面向全院科研管理大数据的数据仓库, 为项目后续在线分析、数据挖掘、搭建知识图谱、学科态势和竞争力分析等需求提供平台支持.

1 研究现状和相关工作

数据仓库(data warehouse)是一个面向主题的(subject oriented)、集成的(integrated)、相对稳定的(non-volatile)、反映历史变化(time variant)的数据集合, 是在现有数据库的基础上, 对其中的数据再次进行抽取、加工和使用, 并最终用于管理决策的集合, 并不是简单的数据复制或数据累加^[3,4]. 数据仓库当前主要的应用场景包括报表展示、实时查询、BI (Business Intelligence)展示、数据分析、数据挖掘、模型训练等方面.它提供了一种有效的访问这些数据的方法, 可以帮助科研机构快速而正确的做出决策.

传统数据仓库大都是基于Oracle、MySQL这样的关系型数据库, 扩展成本高, 面对PB级别的数据量以及各种关系数据库、NoSQL数据库、XML文件等数据源, 其处理速度和处理效率不能够满足数据存储、查询以及融合多维度数据进行分析的需要^[5].

广义上来说, Hadoop大数据平台也可以看做是新一代的数据仓库系统, 它具有很多现代数据仓库的特征, 且具有低成本、高性能、高容错和可扩展等特性, 被企业所广泛使用. IBM的研究人员将基于Hadoop平台的SQL查询系统分为两大类: Database-Hadoop hybrids和Native Hadoop-based systems. 第一类中只是使用了Hadoop的调度和容错机制, 使用关系数据库进行查询^[6]. 第二类则充分利用了Hadoop平台的可扩展性, 主要分为3个小类: 1)基于MapReduce的Hive; 2)基于内存计算框架Spark的Spark SQL; 3)基于shared-nothing架构的大规模并行处理(Massively Parallel Processing, MPP)引擎, 如Impala. 在文献[7]和文献[8]对比分析了最具代表性的Hive、Impala和Spark SQL这3种SQL-on-Hadoop查询引擎, 实验表明3个查询引擎均有各自的优点, 综合来看, Hive的查询结果准确率更高, 更为稳定, 但查询时延较为严重, 适合批处理; Impala查询速度最快, 但系统稳定性有待提高; Spark SQL处理速度处于二者之间, 更适合多并发和流处理场景.

基于Hadoop的多种SQL查询引擎各有优势, 但从稳定性、易用性、兼容性和性能多个方面对比分析, 目前并不存在各方面均最优的SQL引擎. 考虑到项目离线批处理和在线流处理的需求, 而目前较少有兼顾两种需求的数据仓库实施方案, 结合当下较为成熟的开源技术方案, 本文设计并实现了面向科研管理大数据的数据存储系统.

2 大数据仓库设计

传统数据仓库大都只用到结构化数据处理技术, 大数据仓库不仅要处理关系数据库中的结构化数据, 还要处理海量半结构化和非结构化数据, 并为大数据分析提供平台, 需要结合大数据技术设计和构建. 以下分别以相关技术分析选型、集群高可用设计、系统设计阐述大数据仓库设计思路.

2.1 技术选型

Hive是基于Hadoop的数据仓库工具, 可以提供类SQL查询功能, 本质是将SQL查询转换为MapReduce程序. MapReduce框架主要适用于大批量的集群任务, 批量执行导致时效性偏低, 并不适合在线数据处理的场景, 一般用来做数据的离线处理. 使用Hive来做用来做离线数据分析, 比直接用MapReduce程序开发效率更高. 因为大多数的数据仓库应用程序是基于关系数据库现实的, 所以Hive降低了将这些应用程序移植到Hadoop上的障碍^[9].

MapReduce框架及其生态相对较为简单, 对计算机性能的要求也相对较弱, 运行更稳定, 方便搭建及扩充集群, 适合长期后台运行. 但其执行速度慢, 不适合实时性要求较高的查询场景, 在保证系统稳定、减少运维难度的前提下, 融合同样基于Hadoop平台且系统相对稳定的Spark框架是更好的选择, 并且能为在线分析、数据挖掘等提供支持.

Spark是借鉴了MapReduce框架并在其基础上发展起来的, 继承了其分布式计算的优点并改进了MapReduce明显的缺陷. Spark SQL作为Spark生态主要组件之一, 与Hive基于MapReduce进行查询类似, Spark SQL使用Spark作为计算引擎, 在使用时需要处于Spark环境. Spark SQL几乎完全兼容HiveQL语法, 只是Hive特有的一些优化参数及极少用语法不支持.

Hive on Spark是由Cloudera发起, 由Intel、MapR等公司共同参与的开源项目. 它把Spark作为Hive的一个计算引擎, 将Hive查询作为Spark任务提交到Spark集群进行计算. Hive On Spark和Spark SQL只是SQL引擎不同, 并无本质的区别, 都是把SQL查询翻译成分布式可执行的Spark程序. 而Hive on Spark与Hive on MapReduce一样可以使用HiveQL语法. 如果要在数据仓库中使用Spark作为计算引擎, 融入Hive on Spark是更好的选择.

综上所述, Hive on Spark与Hive on MapReduce结合, 可以高效切换计算引擎, 同时提高资源利用率, 降低运维成本.

2.2 高可用性

高可用性, 即HA (High Availability), 指的是通过尽量缩短因日常维护和突发系统崩溃导致的停机时间, 以提高系统和应用的可用性.

分布式系统通常采用主从结构, 即一个主节点, 连接N个从节点. 主节点负责分发任务, 从节点负责执行任务, 当主节点发生故障时, 整个集群都会失效, 这种故障称为单点故障.

HDFS集群的不可用主要包括以下两种情况: 一是主节点主机宕机, 导致集群不可用; 二是计划内的主节点软件或硬件升级, 导致集群在短时间内不可用.

在Hadoop2.0之前, 也有若干技术试图解决单点故障的问题. 如元数据备份、Secondary NameNode、Backup NameNode、Facebook AvatarNode方案^[10]等, 还有若干解决方案, 基本都是依赖外部的HA机制, 譬如DRBD^[11], Linux HA, VMware的FT等等. 但以上方案存在需要手动切换、恢复时间过长、需要引入另一个单点等问题.

为了解决上述问题, Hadoop社区在Hadoop2.X版本中给出了真正意义上的高可用HA方案: Hadoop集群由两个NameNode组成, 一个处于Active状态, 另一个处于Standby状态. Active NameNode对外提供服务, 而Standby NameNode仅同步Active NameNode的状态, 以便能够在它失败时快速进行切换. 其原理如图1所示. 集群通过ZooKeeper进行心跳检测, 通过JournalNode独立进程进行相互通信, 同步NameNode状态.

在生产环境中, 必然要考虑到集群的高可用性, 因此集群需要设置一个主节点备用节点, 在主节点出现故障后能够及时切换到备用节点, 保证集群可用性.

2.3 系统设计

基于以上分析,本文采用基于Hive的MapReduce+Spark双计算引擎混合架构进行大数据仓库系统设计, 满足了项目对于数据仓库高效、高可用和可扩展性的需求. 为更好的管理Hadoop和Spark两个计算集群, 提高集群资源的利用率及集群的计算效率, 采用YARN (Yet Another Resource Negotiator)进行资源管理, 保证了整个系统的稳定性和可靠性, 系统架构如图2所示.

系统将来自不同数据库、互联网、第三方的多源异构数据汇聚到HDFS文件系统, 采用Hive进行管理和索引, 再通过上层计算引擎对数据进行查询分析和计算. 通过YARN进行Hadoop集群和Spark集群的资源分配和管理, 并通过ZooKeeper实现系统中Hadoop、Spark、YARN组件的高可用性, 可按需扩展集群节点进行扩容.

依据计算需求不同, 通过配置或简单命令可以随时切换Hive计算引擎. 在对实时性要求不高或对稳定性要求较高的场景下使用MapReduce引擎; 对实时性有一定要求时使用Spark引擎. 两种引擎均使用HiveQL对数据进行操作, 无需切换开发环境, 可以高效利用集群资源对数据进行抽取、转换, 为机器学习和图计算提供数据源, 系统还可以通过Spark Streaming基于HDFS对数据进行流处理, 为实时流处理提供平台.

3 实现与优化

Hadoop和Spark均依赖于Java, 集群需要安装JDK, 同时使用MapReduce和Spark作为数据仓库的计算引擎, 需要安装配置Hive.

以Spark作为计算引擎时, 需要注意Hive版本对Spark版本的兼容性, 具体对应版本在可以在下载Hive源码时查看pom.xml文件中的spark.version, 或者参考Hive官网^[12]. 默认Spark预发布的版本中有Hive的jar包, 要使用Hive on Spark需要去掉这些Spark访问Hive的jar包, 所以需要重新编译Spark源码. 不同的Spark版本编译命令有所区别, 同样参考Hive官网. 编译Spark源码需要用到Maven, 使用Spark框架还需要用到Scala, Spark从2.X版本开始使用Scala的2.11.X版本. 我们使用MySQL数据库存储Hive元数据. 各资源版本如表1所示.

3.1 集群部署架构

Hadoop集群和Spark集群搭建在5台虚拟机上, 虚拟机具体配置信息见表2. 集群设置有一个主节点, 一个主节点备用节点, 进行任务管理, 三个从节点进行任务执行, 架构如图3所示.

3.2 高可用性实现

集群使用YARN进行资源管理, Spark集群部署为yarn-cluster模式, 通过高性能协调服务Zoo-Keeper和ZKFC (ZK Failover Controller process)组件实现高可用, 具体服务部署如表3所示. 目前已实现Hadoop、YARN、Spark主备节点故障切换. 在主节点进程Namenode (Hadoop)、Resourcemanager (YARN)、Master (Spark)因异常退出后, 备用节点能够及时启用, 继续管理集群.

3.3 Hive负载均衡实现与优化

使用Hive数据仓库有三种连接方式:

(1) Hive的CLI操作方式: bin/hive.

(2) Hive JDBC服务:

nohup bin/hive --service hiveserver2 &

bin/beeline

!connect jdbc:hive2://10.2.4.60:10000

(3) Hive命令, bin/hive-e “HQL语句”或者bin/hive-f SQL文件.

以上连接方式, CLI或者Hive命令的方式仅允许使用HiveQL执行查询、更新等操作, 并且比较笨拙单一. 而HiveServer2(HS2)支持多客户端的并发和认证, 并且允许远程客户端使用多种编程语言如Java、Python向Hive提交请求, 取回结果, 为开放API客户端如JDBC、ODBC提供了更好的支持. HS2使得客户端可以在不启动CLI的情况下对Hive中的数据进行操作, 如可以使用beeline连接HS2执行查询. 当集群中存在多个HS2服务时, 用户可以自行选择具体主机进行连接, 但某台服务器连接数过大时容易造成端口不响应, 服务器故障也会造成无法查询, 使用HAProxy可以最大程度避免这种情况. HAProxy是一款提供高可用性、负载均衡, 基于TCP和HTTP应用的代理软件, 并且具有代理集群状态监控功能. HAProxy通过配置前端(frontend)监听端口和后端(backend)服务端口进行请求转发, 并提供多种负载均衡算法, 适合不同场景下的负载均衡. 当客户端向前端绑定的端口发送请求时, HAProxy根据指定的算法选择可用的后端服务, 并将请求转发.

将HAProxy部署在服务器1上, 并将服务器3、4、5作为Hive Server, 运行HiveServer2服务, 工作原理如图4所示.

HAProxy核心配置如下所示:

bind 0.0.0.0:25005 #HAProxy作为代理绑定的IP, 端口

mode tcp #在第四层进行代理服务

balance leastconn #调度算法

maxconn 1024 #最大连接数

server hive_1 10.2.4.62:10000 check inter 180000 rise

1 fall 2

server hive_2 10.2.4.63:10000 check inter 180000 rise

1 fall 2

server hive_3 10.2.4.64:10000 check inter 180000 rise

1 fall 2

将服务器1上的25005作为前端端口, 每当通过beeline客户端向服务器1上的25005端口发起请求时, HAProxy通过leastconn算法(最少连接数分配)^[13]轮询可用的后端服务, 即轮询hive_1、hive_2、hive_3的10000端口, 10000端口为HS2提供TCP层服务的默认端口. 服务器3–5上需要配置Hive元数据对应信息, 客户端通过10000端口获取元数据信息, 进而查询Hive数据. 集群只需对外提供服务器1的统一前端端口, 最终即可实现通过任意beeline客户端访问服务器1的HAProxy代理, 使用服务器3–5上的HiveServer2服务执行Hive查询. 并充分使用每个Hive Server, 分散压力.

3.4 其他优化

(1) 资源管理器调度优化:

YARN主要由ResourceManager、NodeManager、ApplicationMaster和Container等几个组件构成.

ResourceManager是Master上一个独立运行的进程, 负责集群统一的资源管理、调度、分配等等; NodeManager是Slave上一个独立运行的进程, 负责上报节点的状态; App Master和Container是运行在Slave上的组件, Container是YARN中分配资源的一个单位, 包涵内存、CPU等等资源, YARN以Container为单位分配资源.

ResourceManager内存资源配置, 配置的是资源调度相关:

yarn.scheduler.minimum-allocation-mb分配给AM单个容器可申请的最小内存, 默认1024 MB;

yarn.scheduler.maximum-allocation-mb分配给AM单个容器可申请的最大内存, 默认8192 MB, 由于我们所有的虚拟机都是8 GB内存, 需要留2 GB内存给操作系统, 1 GB内存给Hbase, 因此此处将单个Container内存分配上限设为5 GB, 即5120 MB.

NodeManager的内存资源配置, 配置的是硬件资源相关:

yarn.nodemanager.resource.memory-mb节点最大可用内存, 默认8192 MB, 参考RM设置为5120 MB.

yarn.nodemanager.vmem-pmem-ratio虚拟内存率, 默认2.1.

可以计算节点最大Container数量:

max(Container)=yarn.nodemanager.resource.memory-mb/yarn.scheduler.minimum-allocation-mb=5.

(2) MapReduce调优:

MapReduce程序优化关系到Hive作业的每次提交, 一些特定值的设置会较大影响到MapReduce任务执行效率. Map Task和Reduce Task调优的一个原则就是减少数据的传输量、尽量使用内存、减少磁盘IO的次数、增大任务并行数, 除此之外还要根据自己集群及网络的实际情况来调优.

3.5 测试与分析

(1)高可用性测试:

以YARN为例验证自动故障切换, 在Active节点上kill掉ResourceManager服务, 备用节点能够自动由Standby状态切换为Active, 过程及结果如表4所示.

经过测试, Hadoop、YARN、Spark均可以进行主备节点故障切换.

(2)负载均衡测试:

启动3、4、5节点的HiveServer2服务, 服务器3、4、5分别命名为hive_1、hive_2、hive_3, 集群监控管理界面如图5所示, 当前状态为无客户端请求. 通过beeline客户端连接HAProxy服务器前端服务对应端口, 输入存放元数据的MySQL数据库账号密码, 即可成功通过HiveServer2服务对Hive中的数据进行操作, 如下所示:

[root@slave1 bin]# beeline

Beeline version 3.1.0 by Apache Hive

beeline>!connect jdbc:hive2://10.2.4.60:25005

Connecting to jdbc:hive2://10.2.4.60:25005

Enter username for jdbc:hive2://10.2.4.60:25005:root

Enter password for jdbc:hive2://10.2.4.60:25005:**********

Connected to: Apache Hive (version 3.1.0)

Driver: Hive JDBC (version 3.1.0)

Transaction isolation: TRANSACTION_REPEATABLE_READ

0: jdbc:hive2://10.2.4.60:25005&

当多个beeline客户端请求连接时, HAProxy会自动分配可用的Hive Server, 如图6所示, 可以看到Sessions一栏中当前连接数Cur由全部为0变为hive_1为0、hive_2为1、hive_3为1, 已按照leastconn分配算法成功实现请求分配, 可有效利用Hive Server, 均衡Hive请求.

4 结论与展望

本文根据中国科学院教育科研态势感知服务项目现阶段实际需求, 以数据仓库的高可用性和OLAP为目标, 研究、设计和构建了面向科研管理大数据的Hadoop+Spark双引擎数据仓库, 支持对异构数据高效存储, 提供多种查询分析引擎, 为项目后续需求如数据挖掘、搭建知识图谱、学科态势分析等非实时场景提供了数据存储和计算分析平台.

由于Hive对事务弱支持, 且事务执行速度很慢, 存在诸多限制和不便, 不适合高并发的场景. 目前架构具有较好扩展性, 未来考虑整合Hbase以提升数据仓库查询实时性和对于事务更好的支持, 让平台满足更广泛的应用场景.