并行框架下基于位图索引的多表星型连接算法

解晨光，刘明刚

（1.哈尔滨金融学院 科研处，黑龙江 哈尔滨150030；2.哈尔滨金融学院 计算机系，黑龙江 哈尔滨150030）

0 引 言

人们对MapReduce的研究与实践结果表明，其在海量结构化数据的查询处理方面具有一定性能优势［1，2］，可用于分析在集群环境中的大型数据集，计算各种数据，并从中导出有价值的业务数据。许多研究致力于此，并开发了更高级的框架，Yahoo开发建于Hadoop之上的Pig，将数据处理中的多个MapReduce过程进行了封装，提供了更高层次的抽象，并提供强大的数据变换操作，包括数据表的连接操作；而Facebook则在Hadoop上构建了一个数据仓库的框架Hive，让精通SQL技能的分析师能够运行查询存放在HDFS （Hadoop distributed file system）上的大规模数据集。尽管更高级的框架如Hive、Pig或Cascading等都可以实现数据连接，但面对复杂的数据集时，它们在效率和灵活性上与MapReduce框架相比有一定的差距，因此利用MapReduce编程机制实现连接是解决实际问题的重要选择。星型模型是最常见的多维数据仓库模型，本文研究目标是利用MapReduce分布式编程框架的优势，将分布式缓存技术与位图索引技术相结合，针对SSB 星型存储模型［3］的特点，优化查询处理过程，研究低传输成本的半连接技术，进而提升OLAP的查询性能。

1 相关工作

1.1 MapReduce编程机制

MapReduce分布式编程模型有两个阶段构成成：Map阶段和Reduce阶段。用户通过编写map （）函数和reduce（）函数，来完成分布式的程序设计。针对大量数据，我们需要把数据存储在分布式文件系统中，Hadoop支持多种文件系统，而通常我们将数据存入HDFS中。Hadoop系统将MapReduce任务的输入数据划分成等长的数据分片，一般分片的大小等于HDFS的一个块的大小，并为每一个数据分片构建一个Map任务，运行用户自定义的map （）函数处理分片中的每条记录key1／value1，得到中间结果key2／value2，并依据key2值对中间结果进行聚集并排序存储在本地磁盘上。如果有多个Reduce节点，需要将中间结果按一定的分区函数进行分区，分区的作用是将Map任务产生的中间结果进行分片，以便将同一组的数据交给同一个Reduce任务处理。Reduce任务的输入来自Map 任务key2／value2的输出，调用用户定义的reduce （）函数，将最终产生的key3／value3 作为输出存储在HDFS 上，运行过程如图1。

图1 MapReduce执行流程

Map 任务与Reduce 任务之间的数据流成为混洗（shuffle）。而复杂的任务难以用一趟MapReduce处理过程完成，需要将其拆分成多趟简单的MapReduce任务来处理，可以在MapReduce主控程序 （如main （）函数）中利用循环控制MapReduce作业的循环执行，完成N 次迭代。也可以将多个MapReduce子任务串联起来，将前一个任务的输出，作为后一个任务的输入，自动的顺序执行：MapReduce1 （MapReduce2 （MapReduce3 （MapReduce4。

1.2 连接算法

（1）重分区连接 （repartition Join），它利用MapReduce的自动排序、合并机制将分组数据在reduce端实现连接［7，8］。使用一个单独的MapReduce任务，并可以连接多个数据集。Map 阶段完成从多个数据集中读取所需数据，并指定每个数据的连接值，将连接值作为Map函数的输出键。输出值则包含将在reduce阶段需要被合并的值。在Reduce阶段，一个reduce接收map函数传来的每一个输出键的所有输出值，并将数据划分为多个分区。然后对所有的分区进行笛卡尔积 （Cartersian product）连接运算，并生成全部的结果集。

（2）复制连接 （replication join），复制连接则是在map端实现连接。如果待连接的数据集中有一个数据集足够小，可以完全存放在内存中时，就可以使用全复制连接。通过边数据技术将连接中小的数据集复制到所有的map节点。通常复制连接具体步骤如下：①利用分布式缓存 （Districubted cache）将比较小数据的数据集复制到map节点。②在map任务初始化过程中将这个小数据集加载到一个哈希表中。③用大数据集分片中的记录遍历这个哈希表，并逐一判断是否符合连接条件。④输出符合连接条件的结果。

（3）半连接 （semi－join），半连接是另一个map端连接。当待连接的2个数据集都很大时，我们通常采用重分区连接，重分区连接跨机器的数据传输量非常大，因此会造成一个数据连接的瓶颈［7，9］。半连接就是在map端对一个数据集中不参加连接操作的数据进行过滤，以减少连接运算所需的网络IO 开销。实现方法是，选取一个相对较小的数据集，假设是数据集1，将其参与连接的连接key抽取出来，保存为数据集3，数据集3一般很小，可以放到内存中。在map阶段，使用分布式缓存将数据集3复制到各个任务节点上，然后将数据集2中不在数据集3中的key对应的记录过滤掉，剩下的reduce阶段的工作与重分区连接相同。

以上3种连接技术都有其各自的优缺点，那么如何针对不同的数据集应选择连接算法？图2给出了一个决策树。这个决策树是于文献 ［7］中提出的决策树的改进版本。图2中的决策树可以归纳为：如果数据集中有一个足够小到可以放到Map的内存中，那么Map端的复制连接就足够了。如果每个数据集都很大，但其中一个数据集可以在经过一定条件过滤以后减小到可放入缓存中，则可选择半连接。如果每个数据集都大到无法放入缓存，并且你也无法预处理你的数据那么就需要在Reduce 中使用重分区连接了。

本文在之后的研究中，将提出一种半连接技术，针对SSB存储模型的特点，首先执行预处理，根据查询过滤谓词，生成维表位图索引文件，然后利用分布式缓存技术将其分发到Map节点内存中，利用维表位图索引间的谓词向量演算，过滤连接数据集，进而降低连接所需的传输代价。

1.3 SSB星型模型

图2 连接策略决策树

在线事务处理 （OLTP）与联机分析处理 （OLAP）是数据库系统面对的2 种主要工作负载，MapReduce、HadoopDB和Scope等系统则更适合对大规模数据集合中的OLAP操作。为了便于OLAP 处理，大型的数据集需要建立简明的、面向主题的多维数据模式［10］，而多维数据模型可以是星型模式、雪花型模式或事实星座模式。星型模型数据集包括：①一个大的包含大批并且不含冗余的中心事实表；②一组附属维表，每维一个表。由于限制了每个维都是一个维表，这导致了部分维表存在数据冗余，维表是非规范化的，表中的属性可能存在层次性。雪花型模式将星型模式中的某些维表规范化，将数据进一步分解到附加表中，形成雪花的形状，这种表易于维表的维护并节省内存空间。事实星座模式是针对复杂的应用若干事实表共享维表，形成若干个事实星座。雪花模型虽然在存储上实现了最小化代价存储，但大多情况下，与巨大的事实表表相比，这种节省可以忽略，并且查询计划汇总连接操作较多，大大的增加了数据节点之间的数据复制和传输代价，在MapReduce环境下，考虑的首要问题是复制代价而不是存储代价，简单的计算模型更有利于实现与优化，因此本文研究以星型模型研究为基础，提出基于分布式缓存和位图索引的星型模型查询算法。

TPC－H （TPC benchmark H）是一个基于双事实星座模型的决策支持的基准，在模式上形成雪花状，SSB （star schema benchmark）是TPC－H 标准简化合并后的星型化模型，目前学术界和工业界都比较认可的标准，文献 ［6］中对SSB模型有详细的说明，SSB 将模式清晰地分解为4个维表和一个事实表，消除了TPC－H 雪花状模型带来的复杂查询执行代价，简单的模型使其更加适合于大数据并行计算环境下的数据分布，图3为SSB模型。

本文后面的部分将继续讨论针对SSB 模式特点采取的优化工作，将整个维表优化为一个位图索引文件，通过分布式缓存技术将其同步到所有任务节点，进而将事实表分片与维表的连接操作转换为按事实表外键直接访问内存中位图索引下标位，并通过维表位图索引列间的谓词演算完成维表的过滤操作。

声音传出来，虚弱得像一声蚊嘤。狼剩儿显然是听见了。我看到他似乎颤了一下，好像从梦中惊醒。他瞄一眼手中的拨浪鼓，又看着奄奄一息的我，口唇开始抖动，越来越剧烈，泪珠大颗大颗地滚落下来。他突然又向前跨了一步，跪到床上，抬起我的头枕在他的臂弯。狼剩儿的眼泪洒落在我的脸上，他不停地用手给我擦拭。我拼尽最后一丝气力，哽咽着说：

2 基于位图索引的多表星型连接优化算法

2.1 存储模型

图3 SSB模型

本文研究是以事实表为中心，包含多个维表的星型模型为基础，为了发挥MapReduce分布式计算框架的优势，并减少事实表更新时为维持数据副本同步所产生的代价。存储模型将维表经过处理转化成为位图索引文件，利用分布式缓存技术，将维表转化后的位图索引文件分发到所有任务节点，利用向量间的逻辑运算，简化维表间的连接分组操作。模型如图4所示。

2.2 查询功能分解

需要针对原始的SQL查询，进行分解，将复杂的查询转换符合存储模型的计算模型。下例是SSB模型支持的Q3查询功能分解：

可将上述查询分解为以下几个部分：

连接谓词，对应事实表与维表之间的连接的主键与外键：

维表上的谓词表达式，表示连接的过滤条件：

图4 基于分布式缓存和位图索引的多表星型连接模型

维表分组：

聚集运算：

排序：

通过查询分解，可以看到，维表在SQL 语句中完成的是过滤和分组的作用，本文将根据维表过滤分组条件，将过滤条件优化为位图索引，用一位来表示事实表中的记录是否满足该维表的过滤条件，通过位图索引列的谓词演算，实现事实表数据的过滤，缩减连接数据集依赖的大小，从而将连接优化为事实表按外键直接访问维表位图索引下标位的操作。

2.3 基于位图索引的多表连接算法MDMO－OLAP步骤：

（1）生成位图索引，根据每一个SQL 查询的过滤条件，为每一个维表生成位谓词向量文件，谓词向量文件表示与维表等长的位图索引，每一位置0或1，表示该维表的记录是否满足过滤条件。

（2）利用Hadoop的Distributed Cathe机制将生成的位图索引文件并分发到每个任务节点内存中，为执行并行连接操作做准备。

（3）利用Maprecuce框架，将大事实表分片，分发到Map端，在Map端，每个Map任务从Distributed Cathe中读取位每一维的位图索引文件，进行二进制谓词演算，完成维表上的过滤操作，从而将连接依赖数据集最小化，再与事实表分片进行连接运算，从而减少网络传输代价。图5是简化后的Q3基于位图索引文件的连接操作。

（4）在分片上完成可分布的聚集函数 （sum （）、count（）、max （））以及代数函数运算 （average （）、max＿N（）、min＿N （））所需的可分布函数如 （avgerage＝sum（）／count（））。整体函数则不进行处理，留到Reduce端进行出处理。

（5）对分片连接运算的结果利用哈希函数，用以将满足条件的分组分配到同一个Reduce节点中。

（6）将结果汇总到Reduce节点中进行归并，完成代数聚集函数合并和整体聚集函数 （median （）、mode（）、rank （））的运算，并将结果输出。

3 实验结果分析

我们利用8台PC机进行实验测试，每台计算机配置为Intel Pentium （R）Dual－Core CPU E5300＠2.60GHz 2.60 GHz，4 G 内 存，500 G 硬 盘，操作 系 统 为CentOS6.3，HADOOP版本采用1.2.1，其中一个节点为中心NameNode节点，负责存储并处理全部的维表数据，7个节点为Data Node节点，用于分布式并行处理，每个节点存储事实表分片，通过JAVA 语言开发实验系统。

图5 基于位图索引的连接操作

实验从SSB中选取查询集合作为测试数据集合，利用dbgen生成器生成数据集，表的大小受比例因子SF 影响。其中事实表LINEORDER 行数为SF＊6，000，000；维表PART 行数为200，000＊ （1og2SF）；维表SUPPLIER 行数为SF＊2，000；维表CUSTOMER 行数为SF＊3，000；维表DATE 行数为2556。设置不同的比例因子可以生成不同大小的数据集。SF＝1时生成1G 大小数据集，SF＝10生成10 G 大小数据集以此类推生成1 T 数据集时SF＝1000。本文提及的MDMO－OLAP算法中，需要生成位图索引，由于维表都比较小，几百毫秒即可创建完成，而分发到分布式缓存，进行数据分发过程中，由于位图索引比较小，所以传输代价也十分低，加上查询结果的归并，这三项与算法执行查询的总时间比只占1.5%左右，因此该算法将复杂的多表连接操作，简化为简单的基于内存的连接技术，用极低的代价，保证了数据连接的性能。

图6是针对SF＝10时，产生10GB数据，选取SSB中查询集合作为测试集合。MDMO－OLAP算法与传统的MapReduce的重分区连接算法 （RU）和Hive查询时间的比较，结果表明，改进后的MDMO－OLAP算法，相比于重分区连接算法和Hive提供的查询在执行星型数据连接时执行效率有所提高。图7是针对不同数据集MDMO－OLAP算法运行时间随查询数量的增加呈递增趋势，说明该算法能减少MapReduce任务的执行时间，减少任务执行过程中的IO开销，当数据集越大时，该算法效率越高。

4 结束语

文中针对以事实表为中心SSB 数据模型，利用节点计算机内存，结合MapReduce分布式缓存策略，提出了一种适合SSB数据模型查询处理的存储模型，同时通过位图索引技术，优化表连接，减少了MapReduce框架多表查询处理连接操作过程中的传输代价，从而提高了整体分布式并行OLAP处理的性能，经测试，该算法取得了很好的负载均衡效果。我们将在未来的工作中将本文研究的多表查询优化技术应用到具体的星型模式存储的商业智能系统中（如，某商业银行的信息化数据仓库），进一步提升并行OLAP处理的性能。

图6 查询运行时间结果比较

图7 不同数据规模时间比较

［1］HUANG Shan，WANG Botao，WANG Guoren，et al.A survey on MapReduce optimization technologies［J］.Journal of Frontiers of Computer Science and Technology，2013，7 （10）：865－885（in Chinese）.［黄山，王波涛，王国仁，等.MapReduce优化技术综述［J］.计算机科学与探索，2013，7 （10）：865－885.］

［2］Yang H，Dasdan A，Hsiao RL，et al.Map－reduce－merge：Simplified relational data processing on large clusters ［C］／／Proceedings of the ACM SIGMOD International Conference on Management of data.ACM，2007：1029－1040.

［3］O’Neil P，O’Neil E，Chen X.Star schema benchmark－revision 3 ［R／OL］.USA：University of Massachusetts Boston.http：／／www.cs.umbo edu／－poneil／StarSchemaB.PDF，2009.

［4］Vernica R，Carey MJ，Li Chen.Efficient parallel set－similarity joins using MapReduce ［C］／／Proceedings of the ACM SIGMOD International Conference on Management of Data.NY，USA：ACM，2010：495－506.

［5］SUN Dalie，LI Jianzhong.MapReduce－based Skyline－join processing ［J］.Journal of Harbin Institute of Technology，2012，44 （1）：103－106 （in Chinese）.［孙大烈，李建中.基于MapReduce的Skyline－join查询算法 ［J］.哈尔滨工业大学学报，2012，44 （1）：103－106.］

［6］ZHAO Baoxue，LI Zhanhuai，CHEN Qun，et al.Sharingbased multi－query optimization approach using MapReduce［J］.Application Research of Computers，2013，30 （5）：1405－1409（in Chinese）.［赵保学，李战怀，陈群，等.基于共享的MapReduce多查询优化技术 ［J］.计算机应用研究，2013，30（5）：1405－1409.］

［7］Blanas S，Patel JM，Ercegovac V，et al.A comparison of join algorithms for log processing in MapReduce［C］／／Proceedings of the ACM SIGMOD International Conference on Management of data.ACM，2010：975－986.

［8］Okcan A，Riedewald M.Processing theta－joins using MapReduce［C］／／Proceedings of the ACM SIGMOD International Conference on Management of Data.NY， USA：ACM，2011：949－960.

［9］Afrati FN，Ullman JD.Optimizing multiway joins in a MapReduce environment［J］.IEEE Transactions on Knowledge and Data Engineering，2011，23 （9）：1282－1298.

［10］ZHANG Yansong，JIAO Min，WANG Zhanwei，et al.Onesize－fits－all OLAP technique for big data analysis［J］.Chinese Journal of Computers，2011，34 （10）：1936－1946 （in Chinese）.［张延松，焦敏，王占伟，等.海量数据分析的Onesize－fits－all OLAP技术 ［J］.计算机学报，2011，34 （10）：1936－1946.］