用 mapreduce 处理 Theta-joins

用 mapreduce 处理 Theta joins 摘要 对于许多数据分析任务来说 连接是基本的 job 但是 mapreduce 范式不直接支持 它们 在等值连接方面有进展 一般来说 不能充分理解 mapreduce 里的连接算法的实 现 我们研究这个问题 任何吧任意连接条件 映射到 map 和 reduce 函数 例如 一个只基于 key 等值 的控制数据库的并行的基本结构 我们提出的连接模型简化了 mapreduce 里的连接的创建和 推理 使用这个模型 我们导出一个惊人的简单的随机算 法 叫做 1 bucket theta 这个算法是用来在一个单 mapreduce job 实现任意的连接的 theta joins 该算法仅仅需要最小的概率 输入基数 我们有证据表明 对于各种各样 的连接问题 它或者接近最优的 或者是最佳的可能的选择 对于 1 bucket theta 不是最 佳选择的问题 我们显示通过使用附加的输入概率如果来达到更好地性能 所有的算法被 制作成 memory aware 它们都不需要对 mapreduce 环境做任何的修改 实验显示 我们的方法的高效性 1 前沿 非常大的数据集在许多学科形成一个挑战 英特网公司想分析 TB 级的应用日志和点 击流量数据 科学家必须处理大规模实验和传感器收集的数据集 例如 大型强子碰撞型 加速装置 国家虚拟天文台 零售商想找到顾客和销售数据里的模式 当处理这个分析 并行计算对确保合理的响应时间是必要的 Mapreduce 已经显示作为最流行的用于并行处 理的范式 Mapreduce 在数据管理研究方面已经有很大的影响 Mapreduce 成功的一个主 要的原因是 免费 开源的实现的可用性 hadoop 和一个活跃的开发者团体 这个团体一 直改进 mapreduce 并添加特性 最近 添加了 database inspired high level 语言 PigLatin 和对 SQL 查询的支持 Hive Mapreduce 被设计用来处理单个的输入数据集 因此不直接支持连接 但是 最近 的研究显示 等值连接已经通过使用 mapreduce s key equality 基于数据库管理 实现了 但是在许多应用中 更复杂的连接位置也不要被支持 对于空间数据 band joins 和 空间 连接很常见 数据集间的关联分析也需要相似的连接 甚至 大家很少关注传统的非等值 连接谓词 Map reduce merge 扩展支持各种连接位置 但是它需要对 mapreduce 做基本 的改变 它如何使用 它不仅添加一个新的合并阶段 而且需要用户写代码 这些代码明确的 必须明白这个实现的分布式特性 本人我们提出允许在 mapreduce 里的任意 theta joins 的高效的并行执行 不必要对 mapreduce 环境进行任何的修改 同时用户也不必要写任何的有专门目的的代码来管理数 据流 所有的事情通缩简单的指定合适的 连续的 map 和 reduce 函数来实现 特别的 我们做了如下主要贡献 1 我们提出了一个 reducer centered 代价模型 和一个连接模型 连接模型简单了 mapreduce 里可能的 theta join 的实现的创建和推理 2 我们提出了一个叫做 1 bucket theta 的随机算法 用来计算一个 mapreduce 任务里 的任何的 theta join 包括笛卡尔积 这个算法仅仅需要最小的输入概率 输入集的基数 3 对于一个流行的非等值连接的类 包括非等值和 band joins 我们提出算法 该算法 在 1 bucket theta 上改进 只要充分的详细的输入是有效的 本人剩下的部分组织如下 我们在第二部分研究了 mapreduce 和一种普通的等值实 现 在第三部分 我们提出了一个定性的 mapreduce 代价模型 和一个连接模型 使用这 个连接模型 我们到处 1 bucket theta 算法 同时在第五部分 证明它的性能 第六部分 显示代表性的实验结果 第七部分讨论了相关工作 第八部分总结了本文 2 mapreduce 和 joins 2 1 mapreduce 综述 Mapreduce 被提出用来在分布式并行架构上简化大规模数据处理 尤其是 商业硬件集 群 这个编程模型的主要观点就是隐藏数据分布和负载均衡的细节 让用户注意力集中在 数据处理方面 一个 mapreduce 程序有 2 个基元组成 Map 和 reduce Map 函数被应 用于单条的输入记录为了计算许多中间结果 对于每一个 key reduce 处理带这个 key 的 所有值得列表 任何 reduce 产生的输出写到分布式文件系统里的一个文件中 Mapreduce 构架的综述在图 1 给出 输入记录 可能给分配到分布式文件系统上的多个 物理位置上 一旦 mapreduce 工作被初始化 这些记录被转移到 mapper 节点上 对于每 一个输入记录 执行一个新的 map 的实体 它把记录解析成一个类型 k1 v1 key value 对 然后输入一个新的类型 k2 v2 key value 对 这些 map 的输出在 mapper 节点局部 收集 基于一个函数 类型 k2 的 keys 被分配到 reducer 节点 Mapper 节点对他们的中间 输出进行洗牌来为每一个 reducer 节点创建一列 key value 对 这些列被转移到合适的 reducer 节点上 一旦 map 阶段完成 每一个 reducer 节点对对它的输入根据 key 排序 然 后 每一个 k2 list v2 的类型对的结果调用一个 reduce 函数来处理 Reduce 的输出 类型 list v2 根据以前的 mapreduce 论文 然后被传送给 dfs 节点 注意到 reduce 也能产 生不同类型 list v3 的输出 2 2 等值连接的例子 考虑数据集 S 和 T 的在共同属性 A 上的等值连接 也就是连接条件 S A T A 通常 这 个通过把连接属性作为 key 确信所有相同连接属性值的对在一个单个的 reduce 函数调用 一起处理 更详细 对于数据集 S 的每一个输入元组 S map 输出 key value 对 s A s 注意到 s 也通过添加一个属性原点 表明这个元组来自数据集 S 来扩充 来自数据集 T 的原则相似地处理 对于每一个连接属性值 reduce 然后计算相应的元组 有 S 符号标志 的元组和有 T 符号标志的元组 间的笛卡尔积 这个是在文献中常见的实现 见第七部分 我们因此参考它作为标准的等值连接实现 这种算法有两个问题 第一 reducer 节点的数目受输入数据集的不同的 A 值得个 数的影响 原因是 所有 A 值相同的元组必须被同一个 reduce 调用处理 固有地限制并行 第二个问题是由数据斜交引起的 如果某些 a 值出现次数非常频繁 处理它的 reducer 接 到过分的大量的工作份额 不但处理输入 而且把大量的结果写到 DFS 因此延迟了工作 的完成时间 2 3 非等值连接的例子 考虑到数据集 S 和数据集 T 连接 非等值连接条件是 S A T A 非等值连接对于 mapreduce 来说 有固有的困难 因为数据集 T 中的每个元组必须不但和数据集 S 中有相 同 A 属性值的元组连接 而且要和数据集 S 中有不同 更小 A 属性值的元组连接 基于 计算模式任何把非等值连接条件映射到 key equality 这点还不是很明显 有人可能会考虑到如下的 天真的 方法 假设属性 A 的值域的值是非负整数 为了 确保数据集 T 的一个元组和数据集 S 中所有 A 属性值等于或小于该元组 A 值的元组进行连 接 我们使 map 输出的数据集 T 的每一个元组都以较小的 A 属性值作为 keys 更精确地说 对于数据集 S 的输入到 Map 每一个的元组而言 map 仅仅输出 s A s 但是对于数据集 T 中的每个元组 t map 输出 a t a T A 也就是只输出 a 值比 s A 要小于或等于的 T 的 元组 t 输出形式是 a t 但是 这种方法也有两个主要的问题 第一个问题 根据连接属性的值 产生大量的 T 元组的 副本 第二个问题 如果属性值 A 不是整数或者可以等于复制 我们不能穷 举所有比给定 t A 更小的值 针对这些状况 map 函数需要知道数据集 S 的属性 A 不同取 值的集合 来产生正确的 T 副本 3 preliminary 在下面的分析里 我们一般假设所有的 reducer 节点都有近似相同的计算能力 事实上 这点不但对于云中创建的虚拟机集群有效 而且对于运行在商品硬件上的物理集群也有效 3 1 优化目标 对于已知的连接操作和它的输入和已知处理节点的数目 我们想最小化工作完成时间 工作的完成时间包括 mapreduce 的所有阶段 从第一个数据元组被传送到 mapper 节点 到最后一个输出元组写回到 DFS 中 用户需要短的工作完成时间 正如我们上面讨论的 也固有的带来一个 负载均衡方法 这和当前关于分布式并行系统的工作一致 当前的分布 式并行系统中 负载均衡思想起了核心作用 3 2 mapreduce 代价模型 对于一个给定的连接问题 相互矛盾的 mapreduce 实现仅仅在它们的 map 和 reduce 函数有不同 既然把输入从 dfs 传递到 mapper 节点 和每一个 mapper 本地的读输入元组 不受具体的 map 和 reduce 函数的影响 因此 我们在最优化的时候 没必要考虑这些代 价 Map 和 reduce 函数影响从产生 map 函数输出到 把最后的连接结果写回到 dfs 的代价 为了分析这些 mapreduce job 阶段的完成时间 考虑一个单个的 reducer Mapper 输出的子 集作为 reducer 的输入 Ruducer 根据输入元组的 key 对元组进行排序 读一个 key 的相应 的 value list 对这个列表计算连接 然后 把局部产生的连接结果元组写到 dfs 图 2 说 明了这个过程 由于 Mapreduce 的特性 在 mapper 节点之间平衡负载很容易 另一方面 当使用标 准的等值连接算法 可能 对于某些 reducer 接到了更大的工作份额 这将延迟工作的完 成 因此为了是工作完成时间更少 我们需要将任何节点分配到的最大的工作份额最小化 也就是 reducers 间的负载均衡 如图 2 所示 reducer 的一些时间花在任务上 这些任务的持续根据输入的大小 而其 他的根据输出的大小或者根据输入和输出的大小 通常 代价是单调的 也就是输入越大 处理所花地时间也越多 输出大小也类似 我们因此能通过算法的 max reducer input 和 max reducer output 评估连接算法的质量 也就是 任何 reducer 分配的最大输入大小 和 任何 reducer 最大的输出大小 我们在下述情况间区分 如果 reducer input 相关代价控制 工作完成时间 那么我们 说这个连接问题是 input size 控制的 如果 reducer output 相关代价支配工作完成时间 然后连接问题是 output size 控制的 如果既不控制其他的 我们有一个 input output 均衡 问题 注意连接问题种类取决于特殊的选择的连接实现 对于 input size 控制的问题 通 过最小化 max reducer input 来实现最小化 工作完成是加你 对于 output size 控制的问题 我们需要最小化 max reducer output 对于 input output balanced 问题 我们必须两个都 最小化 注意 我们定性的代价模型没有做关于那种类型代价控制的假定 也就是 我们的定 性代价模型包括 这些 网络传输时间 CPU time 或者本地的 I O 时间控制 所有这些 代价随着 input 大小或者 output 大小或者两者的大小的增加而增加 因此不论网络 cpu 或者本地 I O 是否是瓶颈 最小化 input 大小或者最小化 output 大小或者最小化两者的大 小 可以最小化工作完成时间 3 3 theta join model 我们用 连接矩阵 M 给两个数据集 S 和 T 间的连接建立模型 然后使用这个代表在 mapreduce 里创建和推理关于不同的连接实现 图 3 显示一个例子 数据集 和 一种连接 谓词的相应矩阵 对于第 i 行第 j 列 如果 S 的第 i 个元组和 T 的第 j 个元组满足连接条件 那么矩阵实体 M i j 被设置为 true 图片中的阴影 否则 就设置为 false 不涂 因为 任何 theta join 是笛卡尔积的一个子集 这个矩阵能表示任何的连接条件 3 4 把连接矩阵单元格映射到 reducers 我们的目标是 使得每一个连接输出元组恰巧被一个 reducer 输出 也就是每个 reducer 不输出相同的结果元组 以避免昂贵的后处理或者消除副本 因此 给定 r 个 reducers 我们想把每个矩阵单元以 M i j true 的方式映射到恰巧一个 reducer 上 如果 这个矩阵单元被映射到 reducer R 我们也称 reducer R 覆盖一个连接矩阵单元 有很多可能的映射 可以覆盖所有的真值矩阵单元 我们的目的是 找到从连接矩阵 单元到 reducers 的映射 这个映射最小化工作完成时间 因此 我们想找这样的映射 或 者平衡 reducer 输入份额 对于 input size 控制的连接 或者平衡 reducer 输出份额 对 于 out size 控制的连接 或者在两者之间取得折衷 对于 input output 均衡连接 图 4 显示在选择不同映射中的折衷 左边的图是映射 在 mapreduce 里被标准等值连 接实现使用的映射 连接属性值相同的所有元组都被映射到同一个 reducer 中 这样导致 了 reducer 输入和输出负载都不均衡 例如 reducer R2 接受 5 个输入元组 同时创建 6 个输出元组 当 reducer R3 输入 3 个元组输出 2 个元组 其他两幅图对应新的等值连接算法 我们在以前的文献里没见过这些算法 使用我们 的 theta join 实现的公式化 作为从 true 矩阵实体到 reducers 集合的映射 容易想出更多 的算法 中间的图代表一个非常细粒的映射 即使数据集 S 的第五 S5 第六元组 S6 和 数据集 T 的第六个元组 T6 都有相同的连接属性值 R2 产生 S5 和 T6 的连接结果 R1 产生 S6 和 T6 的连接结果 这个例子也说明了斜交是如何高效处理的 例如 把有连接至 7 的 输出元组分成许多小片 更好地输出负载平衡的下降趋势是每一个 reducer 更大的输入大 小 是因为了使得每一个 reducer 能产生希望的结果而复制元组引起的 例如 S 的第二 和第三个元组必须送到所有的 3 个 reducer 同时注意 R2 和 R3 都能产生输出 S2 T2 和 S3 和 T2 因为他们都有相应的输入元组 为了执行 matrix to reducer 映射 和避免 副本输出 算法必须把映射相关的信息传递给每一个 reducer 右边的映射表明我们如何能实现输出和输出的最好 极大的输出 块被有效的打破 而 输入副本保持低 reducer 输入和输出都很均衡 即使映射不仅仅覆盖了 true 单元格 而 且覆盖了这些对连接输出没有贡献的单元格 比如 M 2 1 这也实现了 这些不影响连 接结果 因为 reduce 消除他们 我们新算法代表了这个基本思想的实际实现 当最小化 reducer 输入元组的副本时 平衡输入和输出代价 引理 1 假如一个 reducer 非配给了连接矩阵的 c 个单元格 那么这个 reducer 至少接受 2 输入元组 个 证明 考虑 reducer 从数据集 S 中接受 m 个元组 从 T 中接受 n 个元组 这个 reducer 最多覆盖 m n 个连接矩阵 M 的单元格 因此 为了覆盖 c 个矩阵单元格 m n c 考虑满足 m n c 的所有可能的非负值 m 和 n 都当 m n 时 m n 的和取得最小值 4 1 bucket theta 算法 第二部分的例子显示和 mapreduce 里实现连接的挑战 数据 斜交 和 用 key 相等的方 式 基于数据流控制实现非等值连接的困难 现在 我们介绍 1 bucket theta 算法 这个 歌算法处理这些挑战 提供关于该方法性能的强壮的分析结果 4 1 笛卡尔积的实现 既然笛卡尔积组合 S 的每个元组和 T 的每个元组 相应的连接矩阵的所有单元格都是 true 我们解释 1 bucket theta 执行 matrix to reducer 映射 现实用来计算笛卡尔积是接 近优化的 讨论这些结果如何扩大到 theta joins 的处理 4 1 1 分析结果 我们第一次考虑平衡 reduces 的 输出相关的代价 既然 r 个 reducers 产生 S T 输出 元组 max reducer output 的 下级的繁殖是 S T r 通常 S 表示集合 S 的基数 和引理 1 一起 暗示 max reducer input 的下一级范围是 给我们以下的引理 2 引理 2 笛卡尔积 ST 的任何一个 matrix to reducer 映射 S T r 分别 2 是 max reducer output 和 max reducer input 的 更低的范围 为了匹配 max reducer output 的 lower bound matrix to reducer 映射必须划分矩阵 以致 S T r 个矩阵单元格被映射到 r 个 reducers 注意 如果单元格 M i j 分配给 reducer k 然后 reducer k 接受 S 的第 i 个元组和 T 的第 j 个元组来做连接 因此 S 的第 i 个 元组必须分到每个 reducer 这些 reducer 在矩阵中对应的区域和第 i 行交叉 类似的 T 的 第 j 个元组必须被送到所有的 reducers 这些 reducers 的范围和矩阵 M 的第 j 列交叉 图 4 所示 根据每行 每列分配的 reducers 的不同数目 输入元组可能被复制许多次 正如 下面的理论所示 对于一些特别的情况 我们能用 square based matrix to reducer 映射匹配 两个 lower bounds 引理 1 笛卡尔积 ST 假设 S 和 T 是的倍数 也就是 S cS 和 T cT cS cT是大于 0 的整数 在这些条件下 对于 max reducer output 和 max reducer input 相应的 matrix to reducer 映射匹配 lower bounds 对于其他的例子 这些例子中 S T 和 r 不满足 引理 1 需要的属性 已知 max reducer output 最小化 max reducer input 可以作为整数线性编程问题被公式化 这些问题 一般都是 NP hard 的 因此解决起来时昂贵的 但是 我们总能找到一个解决方案 这个 解决方案接近最优化 低代价 不是一般性 让 S T 我们先考虑一种极端情况 S 要远远小于 T 更精确 S 也就是匹配 lower bounds 的最优化方块的边长度要比连 这暗示 接矩阵要长 因此 lower bound 不紧密 因为矩阵的划分不比 S 个元组还要多 显而 易见 最优化的矩阵划分成 r 个区域将由 S T r 个矩阵组成 引理 2 ST 考虑 matrix to reducer 的映射 这些映射完美的平衡所有的输出 每个 reducer S T r 元组 让 S T r 在这个条件下 通过把矩阵划分成 r 个矩阵的单行 0 盖连接矩阵 S T 个单元格中的 x S T 个 有 2中的一个 max reducer input 值 证明 如果如果矩阵的 x S T 个单元格必须被 r 个区域覆盖 一个区域对应一个 reducer 遵循 pigeonhole 原理 至少有一个 reducer 必须覆盖 x S T 个单元格 或者更多的单元 格 这和引理 1 一起推出至少 2输入元组需要被送到那个 reducer 正如我们在 section4 1 1 所示 1 bucket theta 事实上证明了 max reducer input 值至多是 4 一般非常接近 2 使用不同的 matrix to reducer 映射 1 bucket the 与 其他任何 theta join 算法的 max reducer input 比率 最大是 4 2 2 例如 和任何的其他的连接实现对比 这些连接实现的 matrix to reducer 映射必须覆盖 50 或 者更多的连接矩阵单元格 1 bucket theta 的 max reducer input 至多是那种算法的 3 倍 注意 实际上有一个十分松散的最大值范围 例如 当 100 个 reducer 节点参与工作 输入的大小差不多一样 例如 其中一个至多比其余的大 4 倍 max reducer input 接近 2 5 而不是 4 然后 至少覆盖矩阵 50 的任何映射的最坏情况下的比 率仅仅是 1 25 1 8 0 5 总之 除非 x 非常小 没有其余的 matrix to reducer 映射将导致更低 例如 使用 1 bucket theta 大于 3 的因子 不同的规定 提高 1 bucket theta 的工作 完成时间的唯一方式是找到一个 matrix to reducer 映射 这个映射不给任何的 reducer 分 配百分比非常大的连接矩阵单元格 例如 至少 50 回想起正确性 值是 true 的所有连接矩阵单元格必须分配给 reducer 见 section3 4 这意味着 对于 产生很多笛卡尔积的任何连接 1 bucket theta 也被证明接近最优 根据 max reducer input 和 max reducer output 对于有每一个选择条件的连接 通常比 1 bucket theta 有更低的 max reducer input 的 matrix to reducer 映射存在 为了提高 1 bucket theta 我们必须找到这样的一个映射 因 为引理 3 该映射的必要条件是 仅仅覆盖相对很小的百分比的连接矩阵单元格 例如 少于 50 正如下面的讨论所示 实际上 由于需要输入统计和连接条件找到有这种属性 的映射通常很难 输入统计 仅仅知道 S 和 T 的基数 判断任何矩阵单元格是不是连接输出是做不到的 不妨假设 s t 是 由满足连接条件的一个 S 元组和一个 T 元组组成的对 如果连接算法 假设一些矩阵单元 M i j 不是连接结果的部分 很容易通过新建集合 S 和 T s 和 t 分别 分配到矩阵 M 的第 i 行好第 j 列 创建一个反例 为了确定不需要被覆盖的矩阵单元格 需 要更多详细的输入统计 连接条件如果连接条件是一个用户自定义的黑盒函数 然后 知道我们实际上评价矩 阵单元格的连接条件时 才知道那些连接矩阵单元格的值是 false 但是 这违背算法的目 的 为了找到有效率的连接算法 我们实际上必须计算所有单元格的连接 因为没有任何 reducer 覆盖它们 我们实际上把这些单元格作为候选者 即使连接条件不包含用户自定 义的函数 实际上 确定连接矩阵的大的区域通常是很难的 这个算个不确定整个区域不 包含连接结果元组 5 2 M Bucket I 和 M Bucket O 对输入大小控制的连接 我们想找到这样一个映射 最小化 max reducer input 一般 找到一个所有候选单元格的最优化覆盖 是一困难的问题 因此我们提出一个快速的启发 式方法 因为需要更详细的 输入统计学信息 multiple bucket 直方图 和最小化 max reducer Input 算法 3 显示 M bucket I 的伪代码 给定的期望的区域的数目 r 每个区域的最大的 输入大小 maxInput 和一个连接矩阵 M 该算法把覆盖问题分写成一步步覆盖连接矩 阵的子区域 为了保存相似的子问题 M Bucket I 仅仅考虑矩阵水平的部分 从第一行开 始 尝试覆盖一块连续行的所有候选单元格 然后 从下一行 这一行也没有被覆盖 开 始 重复相同的处理 继续覆盖一块块的连续行直到要么覆盖所有的候选单元格 或者用 尽 r 个区域不能覆盖所有的候选单元格 在 while 循环的每一次执行 M Bucket I 覆盖一块行 如图 4 所示 块里的行数在不 同的算法 4 间变化 在行 rowS开始的所有块 由 i 个行组成 i 1 maxInput 1 算法 4 计算一个核 核定义为块里每个区域覆盖的候选单元格的平均数目 直观地 我们覆盖尽 可能多的候选单元格 从 rowS开始的可能的 maxInput I 块中 选择有最高核的块 现在这个块的所有候选单元 格都被覆盖 在右下块的下一个行该过程继续 我们的算法不考虑多于 maxInput 1 行 为了减少搜索空间 见 for 循环限制 实际上 这工作的非常好 因为 更高 的块通常高 瘦得区域 这些区域有一个低分 给定的 M 的行的独特块的第一行 rowf 和最后一行 rowl M Bucket I 分配候选单元格在图 5 所示的块中一列接一列 通过创建一个新的区 域 这个区域有初始化 maxInput 的输入能力开始 M bucket I 循环访问每一个列 Ci 分 配 Ci 中的所有候选单元格 在 row f 和 row l 之间 给区域 只要输入能力不被超过 当 在列 Ci 添加候选单元格将导致该区域接受的输入超过他的输入限制 创建一个新的区域 ci 和下一步的列分配给这个区域 知道该区域达到输入大小限制等等 一旦这个块所有 的列被覆盖 算法 5 返回新建的覆盖区域的集合 我们使用 M Bucket I 在一个搜索中为了找到最小的 maxinput 值 M bucket i 为它 找到一个至多使用 r 个区域的覆盖 一个 reducer 一个区域 该二元搜索的 maxinput 的上 下限是 S T 和 2 前者很明显 因为我们能覆盖 S 行 T 的整个矩阵 后者从引理 1 和 max reducer output 至少是 number of candidate cells r 事实中推断 回想起 M bucket i 是为最小化 max reducer input 对于输出大小控制的连接 我们 应该最小化 max reducer output 而不是 max reduer input 对于这个问题 我们开发了一 个叫做 m bucket o 的启发式算法 M bucket o 像 m bucket i 一样处理 但是不是输入大 小限制 maxinput 它通过面积限制区域 也就是 区域包含的候选单元格的数目 注意 于 M bucket o 相比 m bucket i 更好地使用了输入直方图 因为 m bucket o 确切的知道从每个桶的每个数据集来的输入元组的数目 另一方面 每个桶实际的输出大 小可以使 0 到这个通的计数的乘积之间的任何值 因此 m bucket i 能可靠地平和输入相 关的代价 用公平的粗粒度的直方图 M bucket O 能显示输出代价不均衡对于非常细粒度 的直方图 例如 每个桶包含 5 个不同属性值的平均值 实验支持这个观测 这部分描述的 m bucket i 算法能用来做任何的 theta 连接 对于任何类型的连接 我 们通过探讨连接矩阵中候选单元格的位置的特性来进一步提高 m bucket i 算法 尤其 对 于等值连接 band joins 和非等值连接 连接矩阵有下面的单调性 如果 cell i j 不 是一个候选单元格 也就是 它的值是 false 那么或者所有的单元格 k l ki lj 或者所有的单元格 k l ki lj 也是 false 不是候选单元格 因此 我们基 于单调性修剪搜素空间很快地找到所有的候选单元格 5 3 全局算法 给定 2 个输入数据集 S 和 T 以及一个连接条件 我们常常在不同的 mapreduce 实现之 间选择 对于等值连接 有标准的算法 section2 2 M Bucket 算法和 1 bucket theta 我们能计算 section5 1 所示的统计信息 使用 m bucket 算法 根据连接条件 我们考虑了所有应用算法 从他们相应的 matrix to reducer 映射 我们 能给每个算法评估 max reducer input 和 max reducer output 然后 我们使用传统的数 据库代价评估 因为接受输入最大和输出最大的 reducer 决定工作完成时间 局部的 reducer 计算直接听从代价分析 包括 cpu 和 I O 代价 对于 DFS 数据转移 我们通过 一个平均延迟和转移时间的 disk like 模型 细节留给将来的工作 5 4 扩展 内存 给定 r 个 reducers 使用前面 section 提出的算法 在连接矩阵中创建 r 个部分 每一部 分分给一个 reducer 有些时间 这些换分因为太大而不适合放在内存中 连接算法会把 数据存入本地硬盘和从本地硬盘读取数据 如果连接实现假设数据适合放在内存在 就不 用 我们可以通过使算法 memory aware 避免这种情况 而不是让 reducer 数目驱动矩 阵的划分 区域不能超过规定的输入大小 给定内存限制 m 1 bucket theta 用边长为 m 2 的正方形覆盖整个矩阵 M bucket I 不对输入大小执行一个二元搜索 但是对于输 入限制 m 立即运行启发式 如果有必要选择超过 r 个区域 M bucket O 类似的扩充 6 实验 我们讨论 连接真实和复杂数据的算法的结果 用 10 台电脑的集群来做实验 运行 hadoop0 20 2 一台电脑作为头结点 而其他 9 台作为工作结点 每台机器的配置 单核 Xeon 2 4G Hz CPU 8MB cache 8 GB 内存 和 2 个 250GB 7 2K RPM 硬盘 所有 的电脑连在同一个 GB 交换机的网络里 总之 因此这个集群有 36 个核 配个核有 2GB 内存 用于 map 和 reduce 任务 每个核允许同时运行一个 map 和一个 reduce 任务 分布式文件系统块的大小设置为 64MB 所有的电脑作为 dfs 的存储结点 我们提出下面数据集的结果 云 这是一个包含船和土地状况的扩展的云报告的真实数据集 有 382 百万条记录 每 天记录有 28 个属性 总数据量为 28 8GB 云 5 1 云 5 2 从云中随机的选出 2 个样本集 每个有 5 百万条记录 这些样本集用于 输出大小控制的连接的实验 合成的 对于固定的这是这是数据集的一对 每个连接输入一个 每个数据集包含 5 百万条记录 每条记录有一个 1 到 1000 之间的整数 每个数据集 另一个数 据集 我们在同一个范围里使用 Zipf 分布 通过选择一个 0 到 1 0 间的值为 是常 用的 Zipf 斜交参数 6 1 1 bucket theta vs 标准 等值连接 表 1 显示了关于 synth 的多个 值得等值连接计算结果 因为所有实验的输出要 比输入大很多 所以