MapReduce下的连接-聚集查询优化算法示例

þ可在www.sciencedirect.com在线获取ScienceDirectCAAITransactions on Intelligence Technology 1（2016）79e89http://www.journals.elsevier.com/caai-transactions-on-intelligence-technology/原创文章MapReduce上基于RPK表的高效连接-聚集查询Zhan Lia，Qi Fengb，Wei Chenc，Tengjiao Wanga，*a中国北京大学b中国自然科学基金委员会c香港中文大学，中国香港2016年5月28日在线发布摘要连接聚集是数据库系统中一种重要的、应用广泛的操作。然而，在大数据环境下，尤其是MapReduce框架下，连接-聚合查询的处理非常耗时。主要的瓶颈包括两个方面：临时数据引起的大量I/O和查询处理过程中不同数据节点之间的通信开销。为了克服这些缺点，我们设计了一个数据结构称为引用主键表（RPK表），存储表之间的主键和外键的关系。在此基础上，提出了一种基于MapReduce框架的连接-聚集查询改进算法。实验-在TPC-H数据集上的实验结果表明，该算法在通信开销和查询响应时间方面优于现有方法.Copyright© 2016 ， 重 庆 理 工 大 学 . 由 Elsevier B.V. 制 作 和 托 管 。 这 是 一 篇 开 放 获 取 的 文 章 ，CC BY-NC-ND 许 可 证（http://creativecommons.org/licenses/by-nc-nd/4.0/）。关键词：联合聚合查询; MapReduce;查询优化; RPK表;通信代价1. 介绍近年来，大数据越来越受到人们的关注。连接-聚合查询返回关于多个表的连接的聚合信息，广泛用于大数据分析 。 例 如 ， 许 多 TPC-H1 查 询 包 含 用 于 性 能 评 估 的joinaggregate操作。然而，在现有系统如Hive[1]和Pig[2]中运行联接聚合查询是耗时的。分布式系统在大规模数据分析中具有强大的优势.MapReduce[3]是分布式系统中一个重要的并行计算框架。它在学术界和工业界得到了广泛的研究和应用。然而，在MapReduce上处理的连接和聚合操作都是耗时的[4，5]。主要表现* 通讯作者。北京大学电子与计算机工程学院互联网研究与工程中心。联系电话：86 755 26035225。电子邮件地址：tjwang@pku.edu.cn（T. Wang）。同行评议由重庆理工大学负责。1www.tpc.org/tpch。瓶颈来自以下几个方面：读取和写入磁盘上的大量临时数据以及不同数据节点之间的沉重通信开销。当数据大小按比例增加时，总体响应时间会增加。考虑TPC-H中的以下查询选择c.name，从客户c，订单o其中c.custkey<$ o.custkey集团通过c.name这个SQL查询使用TPC-H基准模式并计算每个客户接受的订单数。由于join属性和aggregate属性不一样，我们不能在一个MapReduce作业中执行join操作和aggregate操作。执行此查询的传统方法需要两个MapReduce作业。第一个作业加载两个表，在不同的数据节点之间交换元组，对两个数据集执行连接操作，然后将连接结果写入磁盘。然后，第二个作业加载临时联接结果并计算最终聚合结果。整个计算是耗时的，由于沉重的通信成本和I/O成本。当查询包含多个联接时http://dx.doi.org/10.1016/j.trit.2016.03.0082468-2322/Copyright© 2016，重庆理工大学由爱思唯尔公司制作和主持这是一篇基于CC BY-NC- ND许可证的开放获取文章（http://creativecommons.org/licenses/by-nc-nd/4.0/）。80Z. Li等人/CAAI Transactions on Intelligence Technology 1（2016）79e 89在不同的表上，查询性能将继续下降。为了克服上述缺点，我们设计了一种称为引用主键表（RPK表）的数据结构。在RPK表中，我们存储表之间的主键和外键的关系。这种结构不依赖于环境中远程数据的分布和移动。我们还提出了一个新的算法，在MapReduce上 的 连 接 聚 合 查 询 执 行 。 我 们 只 需 要 一 个 单 一 的MapReduce作业的连接聚合查询。该算法的主要性能缺陷是通信开销和I/O开销，通过RPK表的引入，降低了这些开销。第四节的实验验证了我们提出的算法在提高查询响应时间方面的有效性。本文的其余部分组织如下。我们在第2节中回顾了相关工作。在第三节中，我们描述了我们提出的RPK表结构，并解释了我们在MapReduce框架上的算法处理。第4节展示了我们算法的性能。最后，我们在第5节中得出结论。2. 相关工作聚集查询和连接查询在最近的研究工作中得到了广泛的研究。我们分析了在不同情况下优化查询处理的一些相关工作。这些措施分为以下几大类：传统数据库性能改进的优化方法。文献[6，7]提出了几种方法来提高传统数据库的查询性能。它们的方法，如多个查询的并发执行、索引技术和物理数据库设计，在分布式环境中很难实现和维护。它们也不考虑查询执行期间的通信成本。连接划分方法（JPM）和聚集划分方法（APM）[8]是连接-聚集查询的两种并行处理方法.JPM方法需要分别处理聚合和连接操作。APM方法需要向所有处理器广播一个完整的表。这两种方法在MapReduce框架上都是次优的。修改或扩展MapReduce框架以提高性能。一些研究者[11，12]扩展了map-reduce模型，在一个作业中处理连接聚合。一些研究人员[13e15]旨在通过在MapReduce框架之上构建中间件或缓存结构来提高性能。这些都是针对连接策略的优化，需要修改MapReduce的核心框架。另一方面，我们提出的算法更具有通用性，并且可以提高连接-聚集算子的效率。预计算查询结果。一些分析系统存储大量数据用于决策。这些系统的数据更新时间间隔足够长。因此，可以预先计算一些查询结果，[16]《易经》云：“以德为本，以道为本。然而，随着新数据越来越快地产生，数据更新变得越来越频繁。当数据更新时重新计算查询结果变得复杂和耗时。这种方法也不能保证数据密集型应用程序的adhoc查询处理的效率。我们专注于提高联合聚合计算没有预先计算技术。近似查询处理。一些研究人员使用近似方法来减少查询响应时间。这些方法包括样本方法[21]和在线计算[18e20]。它们为每个查询返回一个具有一定误差范围或保证的近似结果。响应时间将减少很多，但他们不提供一个确切的结果。此外，它对随机数据检索的要求使得它很难在分布式系统上实现。我们工作的目标是设计一个算法，在不牺牲准确性的情况下提高连接-聚集查询的性能。3. 我们的方法在分布式系统中，根据存储资源的可用性，将整个数据集划分并放置在不同的节点上。在连接聚合查询执行期间，每个工作者访问来自不同节点的数据分片，然后对主键和外键属性执行连接操作。之后，过滤所需的属性值并计算聚合结果。当数据量很大时，执行可能会变得复杂。大量的时态数据需要在磁盘上读写，不同节点之间的通信开销会大大增加。这将导致查询处理时间增加和系统性能下降。为了获得更好的性能，我们首先设计了一种新的数据结构，它最大限度地减少了存储成本。在此基础上，提出了一种基于MapReduce的优化连接-聚集查询处理算法。3.1. 数据结构RPK表在本小节中，我们将介绍Reference Primary Key表。然后，我们解释了设计它的原因和存储方法。首先，我们在这里给出定义。定义1. 在表的主键和外键关系中包含主键的表被定义为目标表。定义2. 表中包含主键和外键关系中的外键的表被定义为引用表。由于连接聚合查询通常执行 在表的主键和外键的等价连接的基础上，设计了一种RPK表结构。在RPK表中，我们存储●●●●Z. Li等人/CAAI Transactions on Intelligence Technology 1（2016）79e 8981引用表的主键和外键之间的关系。对于每个目标表，我们为其主键创建一个RPK表。这样就可以直接从RPK表中获取目标表和引用表之间的主键和外键映射信息。我们将目标表的主键属性存储在第一列中，随后的列包含引用表的主键属性。 存储在后续列中的那些记录具有与第一列中的映射记录的主键值相同的外键值。当原始数据集更新时，我们只需要更新RPK表的相应门户。对于RPKtable中的每个元组，只有当目标表和引用表中的相应记录更新时才更新。我们使用TPC-H基准来研究和分析我们的工作的性能。TPC-H[22]是一个广泛使用的决策支持基准。它旨在评估业务分析应用程序的功能。在TPC-H模式中，我们需要为每个目标表创建RPK表。TPC-H模式中的客户表和国家表的RPK表在表1和表2中。在表1中，目标表是Customer，引用表是Orders。第一列存储Customer表中每条记录的排序主键值custkey。随后的列包含Orders表中映射记录的主键记录1的主键等于01.对于Orders表，这些主键为002或003的记录在其外键custkey中具有相同的值（001）。这些记录将在连接处理期间与record1连接。表1RPK-客户表和订单表的表。(a)Customer表表2RPK-国家表、客户表和供应商表。(a)国家表国家钥匙名称…记录1001莫高雷…record2002暴风科技…记录3003奥格瑞玛……………记录25025铁炉堡……………(b)Customer表卡斯特基国家钥匙…记录1001001…record2002042…记录3003002……………记录25025091……………(c)供应商表苏普基国家钥匙…记录1001015…record2002001…记录3003025……………记录25025002……………(d)RPK-国家表nationkey c.custkeys.suppkey记录1 001 001，012 002记录2 002 003 004，025记录3 003 014，015，016，018 023…记录25 025 081，005 003…在表2中，目标表为国家，参考表为客户和供应商。第一列存储Nation表中每条记录的排序主键值nationkey。随后的两列包含表Customer和Supplier中映射记录的主键。记录1的主键等于001。对于Customer表，这些主键为001或012的记录在其外键nationkey中具有相同的值（001）。这些记录将与…(c)RPK-客户c.custkeyo.orderkey…记录1001002,023…record2002008,012,003…记录3003005……………记录25025011,022,033……………在分布式系统中，数据集通过主键水平划分为数据分片然后，每个分割被分配到不同的数据节点，并包含按主键排序的连续记录。每个数据节点包含基于可用资源的多个数据分割。在这种情况下，我们为数据集中的每个目标表生成RPK表。然后，我们水平- tally分区他们使用相同的分区键，这是用来分区原始表。之后，我们将每个分区分配到包含相应数据拆分的同一个节点。利用RPK表中存储的信息，订购键卡斯特基…record1在连接处理期间。记录1001023…record2002001…3.2.我们在MapReduce记录3003001…卡斯特基名称…记录1001洛根…record2002斯科特…记录3003查尔斯……………记录25025Erik……………(b)Orders表…………记录25025094…82Z. Li等人/CAAI Transactions on Intelligence Technology 1（2016）79e 89图1.一、处理查询的传统方法1.在Mapper中处理连接操作，而不扫描引用表。然后在每个Reducer中分别处理聚合操作。在接下来的章节中，我们描述了我们的算法在MapReduce上的详细过程。由于MapReduce已经在学术界和工业界得到了广泛的研究和应用，然而，该算法不仅是MapReduce框架所独有的。它可以应用于不同的系统。3.2.1. 目标表当过滤操作和聚集操作是对目标表中的属性进行的或聚集操作是无效的时，可以消除对引用表的扫描。目标表和引用表之间的关系可以从RPK表而不是原始表访问。由于数据分片和相应的RPK表分区都是按主键排序的，我们扫描RPK表和目标表来获得连接结果。由于数据分片及其相应的RPK表分区存储在相同的数据节点中，因此可以消除不同节点之间的通信开销我们只需要在必要时访问属性的必需部分。以查询1为例，我们已经提到，处理这个查询的传统方法需要两个MapReduce作业。第一个作业计算连接结果，第二个作业生成最终聚合结果。图1示出了用于处理查询1的整个方法。这两个作业分别运行。在两个作业之间，需要在磁盘中写入和读取临时数据。整个计算过程需要在数据节点之间传递大量的记录来计算连接结果。因此，总成本昂贵。当使用RPK-table时，我们只需要一个MapReduce作业就可以完成计算。我们可以消除第一个MapReduce作业，并在一个MapReduce作业中计算聚合和连接操作。我们应该提到，通过重写查询[9]，可以将having子句转换为“where“子句约束，因此我们在算法中不考虑having子句。图2展示了我们的al-taxm生成的MapReduce作业。当此查询试图查找图二、我们处理查询1的方法Z. Li等人/CAAI Transactions on Intelligence Technology 1（2016）79e 8983图三.在Mapper游行。我们只需要计算Orders表中属于每个客户的行数。因此，可以消除订单表的扫描。我们可以通过扫描相应的RPK表来获得Customer表的引用记录数。如图3所示，在Mapper中通过扫描Customer表和相应RPK表分区的每个分割来处理连接操作。这些RPK表与Orders表相比要小得多。同时，数据分割及其相关的RPK表存储在本地。因此，通信成本仅发生在混洗阶段。在shuffle阶段，我们对连接的结果进行分区，在本例中，这些结果包含属性c.name和o.custkey，然后将它们传输到每个Reducer。分区键是此查询中的客户名称。最后，我们使用Reducer中的哈希表[10]计算聚合结果，如图所示。 四、3.2.2. 引用表上的预测和聚合操作当过滤操作和聚集操作是针对引用表中的属性时，需要原始数据记录来计算最终结果。假设我们在查询1中使用MAX（orderdate）而不是MAX（*），我们需要属性orderdate来计算聚合结果。在这种情况下，我们不能排除从引用表访问属性.但是，我们可以使用RPK表通过在查询计划中下推group-by子句来访问属性的有用部分。因此，我们只使用一个MapReduce作业来完成整个查询计算。在Mapper中，我们仍然扫描RPK表而不是引用表。我们使用目标表和RPK表来计算连接结果。在Reducer中，我们需要获取所需的属性（orderdate）来计算最终结果。由于我们在连接结果中有每个记录的主键，因此我们可以从远程节点访问整个记录或仅访问记录的必要部分。节点之间的数据移动仍然小于普通计算。MapReduce上的查询处理算法如下所示。算法1：输入：查询Q，目标表T，RPK表RT2：输出：Q的结果3：函数map（）：4：对于T的记录r，第五章：访问RPK表中主键被r引用的记录6：如果目标表上只有预测和聚合操作，则7：存储记录的主键8：其他9：通过主键获取所需的属性10：在这些属性11：按顺序存储所需的属性12：如果13：emit（group-by键，必需值）14：end for十五：16：按分组键随机播放17：18：函数reduce（key，list r>）19：如果需要，<20：必要时与其他节点通信21：使用哈希表计算聚合结果22：使用部件聚合值计算最终结果见图4。减速器中的游行84Z. Li等人/CAAI Transactions on Intelligence Technology 1（2016）79e 89图五、复杂查询在TPC-H中包含多个连接操作至 于 更 复 杂 的 查 询 ， 如 图 5 所 示 ， 我 们 可 以 使 用ChainMapper来处理这个查询，它允许在单个MapReduce作业中使用多个Mapper我们在三个表上有两个连接操作。因此，我们需要两个映射器和一个还原器来完成这项工作。在第一个映射器中，我们首先扫描在整个执行过程中没有作为引用表执行的表。然后，每个后续映射器在RPK表和前一个结果的帮助下处理一个连接操作。最后在reducer中生成查询结果。因此，我们需要在每个映射器之间传输数据，但我们只需要传输所需的属性和主键值。当我们需要其他属性来完成计算时，我们可以通过记录的主键从远程节点访问它们。尽可能地降低通信成本。对于包含多个group-by键的查询，我们根据作业中处理的表的顺序改变group-by子句中属性的顺序。最后，我们可以在Reducer中使用哈希表计算聚合结果。4. 实验在本节中，我们进行了两个实验，以验证我们提出的结构和算法的有效性。我们在TPC-H基准测试中，通过连接聚集查询，将该算法与Hive和Pig进行了比较。4.1. 实验设置我们在一个有20个节点的集群上运行我们的实验。我们使用19个节点作为从节点，1个节点作为主节点。详细系统见表3。我们使用Hadoop-1.2.1，Hive-0.10.0和Pig-0.7.0。对于我们的实验，我们使用TPCH工具“dbgen“生成500GB的数据然后生成TPC-H数据集中每个目标表的RPK表。TPC-H数据集的结构见图7。对于每个RPK表，我们只需要扫描目标表和引用表一次。生成每个RPK表只需要几分钟。之后，我们使用用于对每个目标表进行分区的相同分区键对它们进行水平分区。然后，我们将每个RPK表分区分配到包含相应数据分裂的数据节点。我们使用五种不同的查询来进行性能分析。使用的查询是TPC-H基准中的Q3、Q10、Q12、Q14、Q19，如表4所示。这些查询之间的差异在于连接操作的数量、聚合操作的类型、表大小和过滤速率。4.2. RPK表的存储成本我们在图8中给出了不同数据大小的RPK表的大小。该结构的存储成本仅为原始数据集的10%左右。由于结构存储成本小见图6。ChainMapper用于复杂查询。Z. Li等人/CAAI Transactions on Intelligence Technology 1（2016）79e 8985表3系统设置。CPU话务员系统存储器磁盘硕士4四核AMD皓龙TM处理器2378 2.4 GHzUbuntu 12.04 LTS32 GB1 TB奴隶4四核AMD皓龙TM处理器2378 2.4 GHzUbuntu 12.04 LTS32 GB500 GB图第七章TPC-H数据集的结构足够了，我们把RPK表加载到主存储器中，从而可以直接使用该结构。4.3. 节点间通信代价为了比较我们提出的算法，JPM方法和APM方法之间的节点间通信成本，我们计算在查询处理过程中交换的数据的大小。结果如图所示。第九章在我们的算法中，大多数连接操作都是在本地完成的在shuffle阶段，我们只传输主键值和Reducer中聚合计算所需的属性。从图9中我们可以知道，我们的方法中的总通信成本比其他方法少得多。表4在实验中。查询函数Q3发货优先级查询客户、订单、行项4.4. RPK表的更新成本当数据集中的记录被更新、更改或删除时，我们需要更新RPK表中的相关数据。因此，当日期更新时，这可能需要更多时间。因此，我们比较的时间成本时，数据更新与不RPK表。结果如图所示。 10.我们可以知道，在两种情况下的时间开销是相似的。使用RPK-table时不会造成太多额外的时间Q10退货报告查询Q12发货方式及订单优先级查询客户，订单，线路项，国家订单，线路项Q14促销效果查询行项目，物料Q19贴现收入查询行项目，部分图8。RPK-表大小，单位为GB。86Z. Li等人/CAAI Transactions on Intelligence Technology 1（2016）79e 89图第九章交换的数据大小（MB）。见图10。数据更新费用。4.5. 查询执行将该算法与Hive、Pig、JPM方法和APM方法进行了性能比较。我们在图中比较了这五个查询所消耗的时间。十一十五。Hive、Pig和JPM方法都使用多个Map- Reduce作业来完成整个查询执行。它们需要一个MapReduce作业用于每个连接操作，一个MapReduce作业用于聚合操作。每个节点必须在连续作业之间通信部分结果、写入和读取临时数据。APM方法只需要Q10、Q12、Q14和Q19的一个作业。但是它需要广播和扫描每个Reducer中不包含group-by属性的表。而且它仍然需要多个作业来处理像Q3这样的查询，其中group-by属性来自多个表。我们的算法只需要一个MapReduce作业来完成每个查询，并尽可能消除通信原始记录。减少了不同节点之间的通信开销，降低了每个作业的计算开销。因此，减少了所消耗的总时间。基于图6所示的结果，我们的算法在总执行时间方面比其他方法更快。这种改进的主要原因包括在HDFS上读取和写入的数据更少，产生的作业更少，作业设置时间更少，检查点成本和通信开销更少。实验结果表明，该算法在MapReduce的连接-聚集查询中具有良好的性能.正如我们上面提到的，我们的方法不仅适用于MapReduce框架。在其他云架构中，我们仍然可以为每个目标表生成RPK表，并使用它来消除对原始数据集的扫描。见图11。疑问3.Z. Li等人/CAAI Transactions on Intelligence Technology 1（2016）79e 8987见图12。查询10.图十三.疑问12.5. 结论本文研究和分析了分布式环境下的连接-聚集查询。我们设计了一种低空间开销的数据结构，称为RPK表。该结构在目标表和引用表的主键和外键上然后我们提出了一种基于MapReduce的聚合查询算法，用于处理大规模数据上包含连接操作的聚合查询。我们的方法提高了查询执行的性能，减少通信开销，减少作业设置时间和磁盘I/O。实验证实，我们的算法具有更好的性能比现有的方法。88Z. Li等人/CAAI Transactions on Intelligence Technology 1（2016）79e 89图14.疑问14.图15.疑问19.引用[1] A. Thusoo，J.S. Sarma，N. Jain等人，Proc. VLDB Endow. 2（2）（2009）1626e 1629。[2] C. 奥 尔 斯 顿 湾 里 德 大 学 Srivastava 等 人 ， Pig Latin ： a not-so-foreignlanguage for data processing ， in ： Proceedings of the 2008ACM SIG-MOD International Conference on Management of Data ，ACM，2008，pp. 1099和1110。[3] J. Dean，S.格玛瓦特通讯社 ACM 51（1）（2008）107e 113.[4] K.H.作者：Lee，H. Choi等人，AcM sIGMoD Rec.40（4）（2012）11e 20.[5] C. 多尔凯尔湾Nrvg，VLDB J. 23（3）（2014）355e 380。[6] M. Muralikrishna，VLDB 92（1992）91e 102.[7] 作案手法Akinde，M.H. Bohlen，Efficient computation of subqueriesincomplex OLAP，in：Data Engineering，2003.诉讼第19届国际会议。IEEE，2003，pp. 163和174.[8] D. Taniar ， R.B.N.Tan ， C.H.C.Leung 等 人 ， INF. Sci. 168 （ 1 ）（2004）25e 50。[9]杨文波，P.B. Larson，VLDB 95（1995）345e 357.[10] Y. Yu ， P.K. Gunda ， M. Isard ， Distributed aggregation for data-parallelcomputing：interfaces and implementations，in：ProceedingsoftheACMSIGOPS22ndSymposiumonOperatingSystemsPrinciples，ACM，2009，pp. 247和260。[11] S.Y. 陈 光 诚 Chen ， An efficient join query processing based onMJRframework，in：Software Engineering，Artificial Intelligence，Networkingand Parallel&Distributed Computing（SNPD），2012 13thACISInternational Conference on. IEEE，2012，pp. 698和703。Z. Li等人/CAAI Transactions on Intelligence Technology 1（2016）79e 8989[12] H. Yang，杨树A. Dasdan，R.L. Hsiao等人，Map-reduce-merge：simplifiedrelational data processing on large clusters，in：Proceedingsof the 2007ACM SIGMOD International Conference on Management ofData，ACM，2007，pp. 1029和1040。[13] S. Zhang，J.汉，Z.Liu等人， 加速MapReduce与分布式内存缓存，在：并行和分布式系统（ICPADS），2009年第15届国际会议。IEEE，2009年，第 472和478。[14] M. Zaharia ， A.Konwinski ， A.D. 约 瑟 夫 等 人 ， OSDI 8 （ 4 ）（2008）7.[15] D. 彭 ， K. 段 湖 ， 澳 - 地 Xie ， Int.J.Database Theory&Appl.6 （ 1 ）（2013）.[16] D. Xin，J. Han，X. Li等人，Star-cubing：computing iceberg cubesbytop-down and bottom-up integration ， in ： Proceedings of the29thInternational Conference on Very Large Data Bases ， Vol. 29 ，VLDBEndowment，2003，pp. 476和487。[17] M.C. Hung，M.L. Huang，D.L. Yang等人，信息科学177（6）（2007）1333e 1348。[18] S.阿加瓦尔湾Mozafari，A. Panda等人，BlinkDB：在非常大的数据上具有有界错误和有界响应时间的查询，在：第八届ACM欧洲计算机系统会议论文集，ACM，2013年，pp. 29E 42.[19] N. Pansare，V.R.博卡角Jermaine等人，Proc. VLDB Endow. 4（11）（2011）1135e 1145。[20]X . Han，J.Li，H.Gao，Inf.Sci. 278（2014）773e 792。[21] N.拉普捷夫湾Zeng C.，中国茶苔属Zaniolo，Proc. VLDB Endow. 5（10）（2012）1028和1039。[22] TPC-H基准测试。http://www.tpc.org/tpch/spec/tpch2.14.4.pdf网站。