Batch Processing (3-1) - MapReduce Reduce-Side Joins and Grouping

Reduce-Side Joins and Grouping

当 MapReuce Job 执行时，它会读取所有的输入资料，相较於资料库来说等於 全表扫描 (full table scan)，在资料库中，如果你查询的资料量不大，资料库通常会使用 index 快速定位至你的资料所在 (2020 Day 8)，如果查询涉及 join，资料库可能会查找多个 index，然而 MapReduce 中没有 index 这种概念的东西 - 至少不是一般认知中的 index，所以若你想在 MapReduce Job 中过滤出少量的资料，其效能会远低於使用 index 查找。

在分析查询的场景下，大部份是从大量资料中做某种程度的聚合，故全表扫描是合理的，再加上 MapReduce 能并行的在多台机器执行，所以我们在谈论批次处理中的 join 时，意味着会读取所有相关的资料集做事，那我们这就来看看 MapReduce join 的细节吧！

例子：分析 user 活动事件

如下图 10-2，左边是 user 的活动事件，但我们想在分析时加入年龄维度，所以需要 join 右边的 user 资料，找到生日。

Srot-merge joins

回忆一下 mapper 做的事，它会从每一笔 record 中提取 key 和 value 出来，因为要做 join，所以我们会有 2 个 mapper ，2 个 mapper 的 key 都是 user ID，整个 join 流程如下图 10-3：

mapper 会以 key 为基准，完成资料的分区及排序，所以相同 user ID 的资料会相邻的当做 reducer 的输入，除了 key 的排序外，MapReduce Job 可以安排 user database 的资料在前面，接着後面的活动 event 再用 timestamp 做排序，这个技巧也称为 二次排序 (secondary sort)。

感谢二次排序，reducer 在处理时，它一次只需要保留第一笔的 user 生日在记忆体中，尔後直接附加生日年份到 event 中，这个算法也称之为 sort-merge join：mapper 输出用 key 排序的资料，然後 reducer 将再合并双方已排序的资料（做 join）。

Group by

除了 join，另一个常在分析用到的功能就是 group by 了，像昨天分析 log 的例子就是以 URL 做为 key， group by 後计算次数，所以，MapReduce 帮我们在 mapper 阶段做好 group by 了。

处理倾斜

看到倾斜 2 字就该警觉一下了，因为 MapReduce 的模式是，把所有相同 key 的资料都带往同一个地方，若有几个 key 的资料特别多呢？以社群网路举例，你我的跟随者可能才 100 人不到，但名人的跟随者可能是上百万，这种不成比例的资料也称为 hot keys。

当有单一 reducer 因为 hot key 的关系，比其他 reducer 收到更大量资料，称为 倾斜 (skew) 或者 hot spots，MapReduce Job 只有在所有 mapper 和 reducer 都完成时才会完成，所以往後 workflow 的 job 都会因为某个 reducer 倾斜的关系，拖慢速度。

当 join 包含 hot keys 时，这里介绍 2 个不同工具的解决办法。

Pig 的 skewed join 方法是，它首先会执行一个简单的 job 检测哪个 key hot，当执行到 join 时，mapper 会把 hot key 中的资料以随机方式送到多个 reducer 中，其他 join 的资料，皆会复制一份到所有分配到 hot key 资料的 reducer 中。

而 Hive 则是以不同方式优化 skewed join，它需要在 table 的 metadata 中明确指定 hot keys，然後把这些资料分散到不同的档案中，当执行到 join 时，它使用 map-side join 处理 (Day 26 介绍)。

今天讲 reduce-side join，明天就讲 map-side join 啦。