针对海量信息处理对传统的商业智能软件提出的挑战，设计并实现了一种基于并行机制的商务智能(BI)系统(BI-PaaS)。该系统搭建在中国移动大云基础设施之上，以云计算的强大并行计算和分布式存储能力为支撑，将ETL、DM、OLAP、Report等各类BI能力并行化。实验结果表明，基于并行机制的BI-PaaS系统可以大大提高BI能力的数据处理性能，有效地支持电信运营的海量数据分析。
0 引言
    随着业务的快速发展、用户的持续增加和移动互联网的兴起，中国移动正面临着数据的爆炸性增长。如何快速地处理海量数据，及时有效地从海量数据中提取有价值的信息，是中国移动的商务智能(Business Intelligence，BI)系统亟须解决的问题。
    BI系统从企业生产系统中提取出有用数据并进行清洗，然后经过抽取(Extraction)、转换(Transformation)和装载(Load)，即ETL过程，将数据合并到一个企业级的数据仓库里。在此基础上。利用合适的查询和分析工具、数据挖掘(Data Mining，DM)工具、在线分析(OnLine Analytical Processing，OLAP)工具等对其进行分析和处理，最后形成知识，支持企业决策。
    数据的海量增长，处理运算的日益复杂，使传统BI系统面临越来越大的压力，难以满足需求，主要体现为以下两点：
    1)计算性能低。
    传统的BI系统，其BI能力的实现往往采用基于内存的串行机制。即将数据全部或批量导入内存中，再依次进行处理。其处理性能受限于单台机器的内存容量和计算能力，无法支持海量数据的分析处理。
    2)可扩展性差。
    可扩展性是指处理性能随系统规模增长的能力，是设计BI系统所追求的一个重要目标。目前的BI系统缺乏一个设计良好、可扩展性强的架构，系统的效率不是随着计算资源的增加而呈线性增长，当系统到达一定规模时会出现效率降低、执行时间难以预测等问题。
    为满足海量数据的计算要求，保持快速的响应和高性能执行，实现BI算法的并行化是解决上述问题的关键。针对以上问题，本文从现代BI系统的需求出发，研究并实现了一种基于并行机制的商务智能系统BI-PaaS，该系统搭建在中国移动“大云”基础设施之上，实现了并行的ETL、DM、OLAP、Report等各类BI能力，使得BI应用能够满足海量数据处理需求。
1 BI-PaaS
    1.1 BI-PaaS架构
    PaaS(Platform as a Service)是一种基于云计算的服务，将云平台能力进行封装，并提供基于云基础设施的开发和托管环境。典型的PaaS有Force.com和Google App Engine。PaaS为满足应用的便捷开发和高性能执行提供了一种有效的手段。因此，本文将PaaS的设计原理引入BI系统，提出了BI-PaaS系统，该系统由开发套件、BI-PaaS平台两部分组成，体系架构如图1所示。

 
图1 BI-PaaS体系架构
    开发套件由一系列BI任务图形设计器和一个集成器组成。BI任务图形设计器包括ETL设计器、DM设计器、OLAP设计器和Report设计器。各个设计器分别将对应的BI能力以元数据的方式进行封装，并进行图形化展示。应用开发者选择各种BI图形设计器，以拖拉的方式进行相应BI任务开发。各个BI任务开发完成之后，采用集成器把各个BI任务组装集成为BI应用，生成元数据描述文件，并将其部署到BI-PaaS平台。开发套件的引入，及其图形化的操作方式，极大地提高了BI应用的开发效率。
    BI-PaaS平台包括应用层、运维层、引擎层、云化能力层和云资源层。云资源层采用Hadoop构建分布式存储和并行计算环境，为BI-PaaS平台提供支撑；云化能力层提供了ETL、OLAP、DM、Report等各类BI组件的并行实现；引擎层负责ETL、OLAP、DM、Report等BI任务的集成与解析；运维层提供用户注册、应用部署、应用监控、数据隔离、应用执行等基础服务；应用层是外部租户访问BI应用的接口。
    1.2 BI-PaaS特点
    BI-PaaS不同于传统的BI产品，具有以下特点：
    1)高度并行，分布存储。
    BI-PaaS以Hadoop来搭建底层基础设施。Hadoop是一个被设计用来在由普通硬件设备组成的大型集群上执行分布式应用的开源框架，包括两大核心元素：MapReduce和HDFS(Hadoop Distributed File System)。MapReduce是一种并行编程模型，基于此模型可以实现具有良好可扩展性的算法；HDFS是一种分布式文件系统，提供了稳定的数据存储环境。BI-PaaS建立在Hadoop之上，将ETL、DM、OLAP、Report等各类BI能力并行化，以满足海量数据存储、计算和分析的需要。
    2)近似线性的高可扩展性。
    随着系统负荷的变化，BI-PaaS可动态调整存储和计算节点的数量，来满足BI系统的计算需求，保证稳定的响应时间与执行性能。加速比接近线性，具有良好的扩展型和资源利用效率。
    3)开发与运营分离。
    BI-PaaS为开发者提供便捷的开发和部署环境，及高性能和高可扩展的运行环境。BI-PaaS定义了自己支持的应用程序模型，为开发和托管环境提供一个理解应用程序的统一规范。开发环境提供了定制和部署应用程序的基本编程元素。托管环境为所托管的应用提供了可伸缩的计算和存储资源，保证应用的高效执行。

 2 并行机制
    2.1 MapReduce并行模型
    MapReduce是Dean等。在2004年提出的并行框架。在该框架下，数据被抽象为(key，value)的形式，而针对数据的操作被抽象为Map和Reduce两个过程。其中，Map是将一个作业分解成为多个任务，而Reduce就是将这些任务处理的结果汇总，从而获得最终结果。用户需要自定义实现Map和Reduce两个函数。
    如图2所示，一个MapReduce任务将输入数据集分割成相互独立的若干块，以便Map能以完全并发的方式进行。一个MapReduce的任务包括3个阶段：第一个阶段进行数据分割，分割后的每个数据块都包含若干个(key，value)对，并被分发给各节点。第二个阶段进行业务处理，逐行读入数据块中的对，经过处理后得到中间结果。中间结果同样以的形式表示，并按照key值排序、分区，把key值相同的结果合并在一起。第三阶段进行Reduce操作，将各个Map的中间结果复制到Reduce节点，对相同key值的中间结果进行合并得到最终结果，并以(key，value)的形式输出。

 
图2 Map-Reduce编程模型
    2.2 ETL并行实现
    ETL负责数据的抽取、转换和装载，为后续的数据挖掘提供一系列较细粒度的预处理操作。本文基于MapReduce系统地实现了并行ETL，包括但不限于以下操作：属性操作、缺值处理、去重复值、区间化、数据连接。下面举例说明上述几种ETL操作的具体并行化实现过程。
    2.2.1 缺值处理
    原始数据记录中的某些属性的缺值会影响数据挖掘算法对于属性的识别与处理，必须予以填充。缺值处理是针对原始数据记录中的缺值属性用客户指定数据值进行填充。并行缺值处理的实现包括一个Map，无Reduce。主要步骤如下。
    1)预处理：总控节点将待处理数据分割为块，分布式存储到多集群环境中。
    2)Map：
    ①从数据块中逐行读入    ②解析value值，对其中的缺值属性用用户指定值替换，否则保持原始值不变；
    ③将处理结果形成Map的输出(key，value)对，其中key为空文本，value为处理过后的每行文本；
    ④将Map处理后的结果输出到HDFS。
    2.2.2 去重复值
    去重复值是将数据样本中的重复记录进行删除，只保留重复记录中的一条。并行行内去重包括一个Map和一个Reduce。主要步骤如下。
    1)预处理：总控节点将待处理数据分割为块，分布式存储到集群环境中。
    2)Map：
    ①从数据块中逐行读入对，其中key为每行数据的偏移量，value为该行记录的文本形式；
    ②直接将该行记录形成M印的输出(key，value)对，其中key为该行记录，value为空文本。
    3)Reduce：
    ①收集具有相同key值的数据，直接形成Reduce的输出(key，value)对，其中key为整行数据，value为空文本，即可实现去除重复记录；
    ②将结果输出到HDFS。
    2.2.3 属性操作
    属性操作主要包括属性交换、属性删除、属性值添加。现以属性添加为例，说明具体的并行化实现。属性值添加是指针对每一行数据，根据确定的计算公式，对已有属性值进行计算，获得新属性值，并添加到记录的末尾。并行属性添加操作包括一个Map，无Reduce。添加新属性的主要步骤如下。
    1)预处理：总控节点将待处理数据分割为块，分布式存储到集群环境中。
    2)Map：
    ①从数据块中逐行读入(key，value)对，其中key为每行数据的偏移量，value为该行记录的文本形式；
    ②解析value值，获得其中每个属性的值，然后按照计算表达式对已经有属性值进行计算，获得新属性值，写在该行记录的末尾：
    ③将处理结果形成M印的输出(key，value)对，其中key为空文本，value为处理过后的每行文本；
    ④将Map处理后的结果输出到HDFS。
    2.2.4 区间化
    区间化是针对连续性数值型数据，将其规约为某些区间值。例如用户年龄字段，可按照如下规则实现区间化：0～9岁归为类0,10～19岁归为类1，依此类推，90岁以上归为类9。并行区间化包括一个Map，无Reduce。主要步骤如下。
    1)预处理：总控节点将待处理数据分割为块，分布式存储到集群环境中。
    2)Map：
    ①从数据块中逐行读入(key，value)对，其中key为每行数据的偏移量。value为该行记录的文本形式；
    ②解析value值，获得其中所需处理的属性的数值，在区间化映射关系的指导下，将原数值型数据替换为相应的区间类别；
    ③将处理结果形成Map的输出(key，value)对，其中key为空文本，value为处理过后的每行文本；
    ④将Map处理后的结果输出到HDFS。

   2.3 数据挖掘并行实现
    数据挖掘是指从大量的、不完全的、有噪声的、模糊的数据中提取隐含的、未知的、非平凡的及有潜在应用价值的信息或模式。数据挖掘的一个重要功能是数据分类，即将数据映射到预先定义好的群组或类。目前常用的分类方法为基于决策树的方法。
    本文基于Map Reduce机制实现了并行数据挖掘，下面以决策树为例说明数据挖掘的并行化实现。
    用决策树进行分类主要包括两个步骤：第一步是利用训练集生成一棵决策树，建立决策树模型，这个过程实际上是一个从数据中获取知识，进行机器学习的过程；第二步是利用生成的决策树对输入数据进行分类，对输入的记录，从根节点依次测试记录的属性值，直到到达某个叶子节点，从而找到该记录所在的类。
    决策树生成方法是通过自顶向下增长节点实现的，生成过程如图3所示。具体步骤如下：

 
图3 决策树构建过程
    1)遍历设定数据集合，确定决策树当前层每个分支节点分别对应的未分类数据集。
    2)根据每个分支节点分别对应的数据集和候选属性集合，通过分别计算候选属性的信息增益值。选取具有最大信息增益值的数据属住作为该节点对应的决策属性。
    3)根据每个分支节点分别对应的决策属性的属性值，分别生成每个分支节点的决策树分支。
    4)分别判断每个决策树分支节点对应的数据集是否对应分类属性的同一属性值。如果结果为否，则决策树分支下的爷点确定为当前层的下一层的分支节点，并将当前候选属性集合减去当前决策属性，作为该分支的候选属性集合，并将下一层作为当前层返回1)；如果是，则执行5)。
    5)将判断结果为“是”的决策树分支下的节点分别确定为其所在分支的最后一个节点。
    重复1)～5)，依次确定决策树第二层至最后一层各节点对应的数据属性．直到各分支下最下层节点对应的数据集，其中所有记录的分类属性皆为同一属性值，则完成决策树的构建过程。
    在第2)步中。其核心是为每个节点选取能最佳分类当前数据集的属性，成为该节点的决策属性。常用的ID3和C4.5等决策树生成算法，都以每个候选属性的信息增益值来衡量其分类能力。决策树上每个节点选择候选属性中具有最高信息增益值的属性作为最佳分类属性，成为该节点的决策属性。为得到特定候选属性的信息增益值，需要统计待分类数据集中该属性每个属性值对应不同分类类别的频度，进而计算该属性的信息熵以及信息增益值。这是决策树生成算法中最主要的计算阶段。
    因此，为了提高处理速度，对第2)步提供并行化实现(如图4所示)，以并行的方式确定决策树中同一层每个候选节点分别对应的决策属性，即以并行方式计算同一层上每个候选节点的各候选属性的属性值对应设定类别的频度，进而根据所得频度计算该属性的信息增益值；以并行方式确定具有最大信息增益值的数据属性为该候选节点对应的决策属性。

 
图4 确定决策属性并行化过程
    并行过程包括2个Map和2个Reduce，具体描述如下。
    1)Map：
    ①从数据块中逐行读入对，其中key为每行数据的偏移量，value为该行记录的文本形式。
    ②解析value值，根据解析结果形成Map的输出(key，value)对，其中key为分支编号＋候选属性名＋候选属性值＋类别，value为局部频度。即该分块数据子集中所有未分类数据属性的每个属性值对应的频度统计项。
    ③将Map处理后的结果输出到HDFS。
    2)Reduce：
    收集具有相同key值的数据，并行合并相同分支下、具有相同候选属性的不同属性值的频度统计项，并计算每个分支下、每个候选属性的信息熵和信息增益值；Reduce输出的(key，value)对为：key为分支编号，value为属性名+信息增益值。
    3)Map：
    ①对第2)步的Reduce输出文件分块并依次读入每行数据，形成Map的输出(key，value)对，其中key为分支编号，value为属性名+信息增益值；
    ②将Map处理后的结果输出到HDFS。
    4)Reduce：
    收集具有相同key值的数据，并行将具有相同(分支编号)的向量子集，按照信息增益值排序，选择并返回最大信息增益值对应的候选属性作为该分支编号对应节点的决策属性。Reduce输出的(key，value)中，key为分支编号+决策属性名，value为空文本。

3 实验
    3.1 应用实例
    客户流失预测在电信领域是一个常见的应用实例，电信公司根据已有的客户的历史数据，对后期客户流失行为进行预测，以便采取措施来挽留客户。客户流失预测的业务流程如图5所示。

 
图5 “流失预测”业务流程
    客户流失预测使用BI-PaaS的开发套件来开发，应用开发包括ETL开发、DM开发、Report开发和应用集成4个步骤。
    1)使用ETL设计器生成ETL处理流。该流程通过缺值处理、去重复值处理，完成对数据的清洗工作；通过数据区间化处理，完成对数据的转换工作，为数据挖掘算法产生合适的数据格式；通过属性值添加处理，为后续数据挖掘算法准备一个类标签属性。
    2)使用DM设计器生成数据挖掘处理流。该流程把经过ETL预处理的数据分为训练集和测试集两类。首先由具有先验知识的训练集数据训练生成决策树模型，然后使用该模型对测试集数据进行分类预测，以发现流失性强的客户群体。
    3)再利用Report设计器，以拖拽的方式生成报表格式，把数据挖掘得到的客户预测结果以合适的方式进行展现。
    4)最后通过集成器，设计BI应用的业务流程，将前面所设计的ETL任务、DM任务和报表任务进行集成，生成BI应用。
    3.2 实验配置
    实验数据是从中国移动的话单数据中抽样获得，每条记录有49个属性，两份数据大小大约为10 GB和100 GB。实验环境由60个PC节点组成，每个计算节点的硬件环境为4核CPU，8 GB内存，1 TB硬盘，1 GB网络适配器。其中2个PC节点各作为NameNode和JobTracker，其他节点作为DataNode和TaskTracker。实验使用的Hadoop版本是hadoop 0.20.1，blocksize设为64 MB且副本数设为2。
    3.3 实验结果及分析
    分别采用数据量10 GB和100 GB作为实验数据，ETL流程文件和DM流程文件统一设置50个Map和50个Reduce，实验机器分别采用10,30和60个节点，用以研究不同DataNode节点数情况下加速比的性能，实验结果如表1所示。
表1 基于Map Reduce的并行ETL时间对比

 
    1)性能。
    实验结果表明采用10～60个节点就可以以较高性能支持100 GB数据的ETL操作和数据挖掘算法。而在经典商用数据挖掘工具中，由于缺乏可扩展性，一般仅能支持300 MB数据的挖掘。
    2)扩展性。
    本实验采用10节点、30节点、60节点、32节点和64节点规模对并行数据处理和并行数据挖掘算法的扩展性进行测试。
    图6(b)中描述了随着节点数增加的加速比情况。显然，当节点不变时，处理的数据量增加，则加速比接近线性；当数据量不变时，增加节点，加速比接近线性。但是，当节点数量相对需要处理的数据量过多时，加速比反而因为非计算开销而远离理论加速比。实验结果表明，并行数据处理和并行数据挖掘算法具有优秀的扩展能力。

 
图6 数据并行处理时间性能和加速比
4 结语
    BI-PaaS搭建于中国移动大云基础设施之上，以Hadoop的强大并行计算和分布存储能力为支撑，将ETL、DM、OLAP、Report等各类BI能力并行化。从而有效地支持电信运营的海量数据分析，提高电信领域数据分析性能、可扩展性，降低系统平台成本。