基于MapReduce 的数据聚集运算算法
冷芳玲，鲍玉斌，高伟，于戈
(东北大学信息科学与工程学院，沈阳 110004)
摘要：为解决数据仓库中海量数据的处理效率问题，可以采用数据聚集预计算的方法，但是
针对海量级别数据的聚集运算非常耗费计算资源，需要巨大的计算能力和存储能力，因此提
出了一组基于MapReduce 的面向海量数据的数据聚集运算算法，主要包括数据的选择、投
影以及等值连接等，并在此基础上，实现了计数、求和、均值等聚集运算，形成了比较完整
的面向海量数据的聚集运算算法。实验结果表明，该算法充分利用了集群系统的计算能力和
存储能力，极大地提高了海量数据的聚集运算效率和基于聚集运算结果上的数据查询效率。
关键词：数据仓库；聚集运算；MapReduce；联机分析处理
中图分类号：TP 311 文献标识码：A
 0 引言
随着数据采集技术的不断发展和廉价存储容量的不断扩大，收集和存储以指数级增长的
数据资源成为可能。而科学应用的计算和数据需求在最近几年里有了急剧地增长。科学计算
数据量的这种指数级增长已经超过了摩尔定律预测的计算能力和存储空间的增长速度[1-2]。
例如，国家“数字海洋”工程中的数据中心目前需要管理多达几百TB 的通过卫星遥感、观测
台站、海洋浮标等渠道收集的大量资料。这个分析过程需要交互式地进行，即进行在线的聚
集查询。总之，这样的应用具有以下特点：1) 需要大量存储空间管理海量且快速变化的数
据；2) 需要系统具有很好的可伸缩性；3) 需要提供高的数据可用性；4) 需要提供复杂查询
及时的响应能力。一般把这种需要处理大量分布式存储且快速变化数据的计算称为数据密集
型计算（data intensive computing，DIC）。数据密集型计算为学术界和工业界都带来了巨大
的挑战[3]。
数据密集型计算是科学界、工业界、计算机学术界几乎同时提出的研究问题。在国际上，
对于DIC 的研究主要来源于科学与工程计算、电子商务以及互联网3 个领域[4-6]。美国太平
洋西北国家实验室（PNNL）正在从支持科学研究的角度对DIC 开展研究，探索基于新的软
硬件平台的海量数据分析技术。同时，以Amazon 为代表的电子商务公司提供的系统（如
基金项目：国家自然科学基金(61033007)；中央高校基本科研业务费专项资金(N100304005)；国家高技术研
究发展计划(863 计划)(2009AA01Z131)
作者简介：冷芳玲(1978- )，女，讲师，主要研究方向：数据仓库和OLAP，lengfangling@ise.neu.edu.cn
 EC2、S3）已成为DIC 相关研究的基础平台[7-8]。大的互联网公司，如Google 和Yahoo，则
利用自己所具有的海量web 内容和用户访问日志信息，研究有效管理和分析处理这些数据
的技术手段。Google 和Yahoo 公司专门成立了进行数据密集型计算应用的实验室。在学术
界也开展了大量的基于MapReduce[9-10]的数据密集型计算的相关研究[11-12]。
国内也开展了针对大规模数据的研究与开发。奚建清等[13]开展了基于MapReduce 的数
据立方的计算；中科院计算所的宋怀明等[14]开展了大规模数据密集型系统中的去重查询优
化的研究；孙光中等[15]开展了MapReduce 模型的调度及容错机制研究；华东师范大学的海
量计算课题组进行了数据密集型计算数据管理的研究[16]。
进行数据聚集预计算是提升海量级数据仓库联机分析处理性能的主要手段，也是 提高
数据密集型计算应用速度的有效方法之一，它主要通过在查询之前就准备好结果，从而减少
查询的反馈时间，这种策略是OLAP 技术能够快速响应的基础，主要体现在2 个方面。
1) 聚集预运算有效减少了查询的反应时间。一般情况下，联机分析处理所需要的数据
是由基础细粒度数据统计出的聚集数据，如果不经预聚集运算而直接在海量基础数据上进行
查询统计，将极大影响系统效率。通过聚集预计算出需要的聚集信息，从而避免对基础数据
的直接访问，减少了响应时间。
2) 聚集运算减少了对基础数据的重复计算。不同的查询操作可能都需要对相同的部分
基础数据进行相同的计算。例如，“2010 年第1 季度A 地区的出货量”是对A 地区该时间段
内的所有销售记录进行统计汇总，在进行“A 地区2010 年不同季度出货量对比”、“A 地区不
同年份第1 季度的出货量对比”、“不同地区2010 年第1 季度出货量对比”等处理时，都需要
使用该汇总数据。通过聚集预先计算出该汇总数据，从而避免对相关基础数据的重复计算。
通常情况下，要进行聚集运算的数据是以关系型数据的形式存储在文件中的，对于海量
的数据，普通的关系型数据库系统的数据处理效率难以满足人们日益增长的查询实时性需
求，因此，研究关系型海量数据的处理意义重大。
1 MapReduce 在关系型数据上的查询处理
为方便叙述，做以下定义。
1) 设关系模式为R（A1，A2，…，An），它的一个关系设为R。t∈R 表示t 是R 的一个
元组。t[Ai]则表示元组t 中相应于Ai 的一个分量。
2) 若A={Ai1, Ai2, …, Aik}，其中Ai1，Ai2，…，Aik 是A1，A2，…，An 中的一部分，则A
称为属性列或属性组。t[A]={t[Ai1],t[Ai2],…,t[Ain]}表示元组t 在属性列A 上诸分量的结合。A’
则表示{A1，A2，…，An}中去掉{Ai1，Ai2，…，Aik}后剩余的属性组。
3) R 为n 目关系，S 为m 目关系，tr 属于R，ts 属于S，
..
r s t t 称为元组连接(Concatenation)。
它是一个m+n 列的元组，前n 个分量为R 的一个n 元组，后m 个分量为S 中的一个m 元组。
4) 给定一个关系R（X, Z），X 和Z 为属性组。当t[X]=x 时，x 在R 中的像集(images set)
定义为Zx= {t [Z] | t∈R, t[X] =x}。它表示R 中属性组X 上值x 的诸元组在Z 上分量的集合。
1.1 选择运算
选择又称为限制（restriction），它是在关系R 中选择满足给定条件的元组，记作
(R) = {t | t ∈ R ∧ F(t) = true} F δ
其中，F 表示选择条件，它是一个逻辑表达式，取逻辑值“真”或“假”。
如何通过MapReduce 计算框架从海量的数据中选出符合条件的元组呢？具体算法与普
 通数据库系统的选择方法一致，最大的差别在于：通过MapReduce 机制实现的选择算法是
并行执行的过程，因为数据被分割为大小相同的若干个块，这些块被分布到集群系统的不同
数据节点上，各数据节点分别读取本地的数据，从而选出符合条件的元组，充分发挥了集群
系统的计算能力。处理流程为：首先各个数据节点读取本地数据块的一个或者若干个元组，
当读取操作完成以后，系统会调度一个mapper 函数，将读取的结果打包后以参数的形势传
递给mapper 函数；然后由mapper 函数判断该元组是否符合给定条件，将符合条件的元组以
<key, value>对的形式传递给combiner，由combiner 对本地相同key 的value（即元组）进行
合并，最后通过计算key 的哈希值，或者用户以自定义的方式决定将该<key, value>对传递
给哪个reducer，最后由reducer 将各数据节点的<key, value>对进行汇总。
对于待处理的数据，一行内容代表1 个元组的形式被有序地存储在文件中，这些文件必
须被存储在HDFS 分布式文件系统上，任务请求首先从名字节点上获得输入文件的位置信
息，然后根据这些信息判断将计算环境转移到哪些数据节点上，各个数据节点根据自身的存
储情况，从文件的本地组成块中每次读取1 行或者若干行的内容，然后将它以参数的形式传
递给mapper 函数处理，mapper 函数检查传递进来的元组是否符合给定条件F，将符合条件
的元组继续向下传递，否则被抛弃不被处理。基于MapReduce 的选择运算如下。
算法1 基于MapReduce 的选择运算
输入：key 为元组所在文件的偏移地址，value 表示元组。
输出：<key, value>，其中key 根据需要用户自定，value 为满足条件的元组。
mapper(key, values)
1) 对value 进行处理，识别出所有的属性。
2) if F(value) = true then
3) Context.write(key, value);//将符合条件的元组存储到本地中间文件中。
4) end if
上述假设每次读取1 个元组，然后传递给mapper 函数，其实为了真正在应用该算法进
行运算时，每次要读取多条元组，这样有助于提高计算的效率。对于这样一个简单的选择运
算，如果1 个任务只完成选择运算这样1 个任务，那么只需要1 个mapper，这样，1 个任务
不需要reducer。通常情况下，选择运算只作为1 个任务的组成部分出现。
1.2 投影运算
关系R 上的投影是从R 中选择出若干属性列组成新的关系。记作
π A(R) = {t[A]|t ∈ R}
其中，A 为R 中的属性列。在MapReduce 计算框架中，这里投影运算的mapper 算法与选择
运算中的mapper 稍有不同，相同之处在于所有存放了关系表内容的数据节点每次仍然读取
1 个元组，投影运算直接在map 阶段就全部完成，所以不需要reduce 过程。通常情况下，
投影运算作为一个查询的组成部分，要融合到相应的map 函数中，不独立操作运算。基于
MapReduce 的投影运算如下。
算法2 基于MapReduce 的投影运算
输入：key 为元组所在文件的偏移地址，value 表示元组。
输出：<key, value>，其中key 保留属性的组合，value 为空。
mapper(key, values)
1) 对value 进行处理，识别出所有的属性。
2) 选出需要的属性，从组成新的key。
 3) value=null。
4) Context.write(key, value)；//将符合条件的元组存储到本地中间文件中。
1.3 等值连接运算
等值连接是从关系R 和关系S 的笛卡儿积中选取A、B 属性值相等的元组，记作
{.. | [ ] [ ]} A B r s r s r s
R S tt t R t S t A t B
=
∞ = ∈ ∧ ∈ ∧ =
其中，A、B 分别为R、S 中的属性列。
在MapReduce 中，首先读取R 关系中的一个元组，将其传递给mapper，传递给mapper
函数的参数key 可以为任意值，通常是所读取内容的偏移地址，当然用户可以根据需要自行
设定，mapper 的参数value 包含1 条或者若干条R 关系的元组，在mapper 中，提取该元组
的tr[A]，将tr[A]作为mapper 输出的key，mapper 输出的value 与输入value 相同；同理，读
取S 关系中的1 个元组，将该元组传递给mapper，mapper 从该元组中提取ts[B]，将ts[B]作
为mapper 输出的key，value 不变。在MapReduce 运行机制中，会将mapper 输出的相同key
学术论文网Tag：代写代发论文 论文发表 职称论文发表 代写机械论文