Question

Apache hive 中的 map join 就是之前提到的 Map 端的 join 或者叫广播 join。还有一种 join 是 common join 或者 sort merge join。common join 和 sort merge join 在执行的时候消耗的资源比较多，运行较慢。为了提高 join 的运行效率，我们可能需要用到 hive 中的 map join。使用 map join 的前提是两个表做关联时需要有一个表是可以加载到内存的小表。这样 join 可以在一个 mapper 里面完成，而不需要 reduce。

尽管查询经常依赖于小表连接，但是 map join 的使用加快了查询的执行速度。

在 MapReduce 任务中，第一步就是创建一个 MapReduce 本地任务，然后该 map / reduce 任务从 HDFS 读取小表的数据，然后把数据保存到内存中的哈希表，然后再存储到一个哈希表文件。接着，当 MapReduce join 任务启动的时候，它会把哈希表文件移到 Hadoop 分布式内存中，并把哈希表文件存储到每个 mapper 的本地磁盘上。所有的 mapper 都能把这个哈希表文件加载到内存，然后在 map 阶段做 join 操作。

用一个例子来理解一下上面的过程，假如有两个表，一个大表 A 和一个小表 B。对于每个大表 A 的 mapper 来说，table B 都能完全读取。因为小表已经被加载到内存了。然后，join 会在 MapReduce 作业的 map 阶段被执行，而不需要 reducer 任务，即 reducer 任务会被跳过不执行。因此，hive 中的 map join 比普通 join 要高效。

Map join 相关参数

下面详细讨论一下 Map join 涉及的相关配置参数。

hive.auto.convert.join

该参数表示是否自动把任务转为 map join。默认该配置为 true。当一个表的大小小于 25 MB（配置参数：hive.mapjoin.smalltable.filesize）时，表之间的 join 会被自动转成 map join。

hive.auto.convert.join.noconditionaltask

是否将多个 map join 合并为一个。默认为 true。多个 mapjoin 转换成 1 个时，所有小表的文件大小总和的最大值由 hive.auto.convert.join.noconditionaltask.size 配置项控制。

例如，一个大表顺序关联3个小表 a(10M)，b(8M)，c(12M)，如果hive.auto.convert.join.noconditionaltask.size 的值：

1. 小于18M，则无法合并 mapjoin，必须执行3个 mapjoin。
2. 大于18M小于30M，则可以合并 a 和 b 表的 mapjoin，所以只需要执行2个 mapjoin。
3. 大于30M，则可以将3个 mapjoin 都合并为1个。

为什么要合并 mapjoin 呢？因为每个 mapjoin 都要执行一次 map，需要读写一次数据，所以多个 mapjoin 就要做多次的数据读写，合并 mapjoin 后只用读写一次，这样就能大大加快速度。但是执行 map 是内存大小是有限制的，在一次 map 里对多个小表做 mapjoin 就必须把多个小表都加入内存，为了防止内存溢出，所以加了 hive.auto.convert.join.noconditionaltask.size 参数来做限制。不过，这个值只是限制输入的表文件的大小，并不代表实际 mapjoin 时 hashtable 的大小。

hive.mapjoin.localtask.max.memory.usage

将小表转成 hashtable 的本地任务的最大内存使用率,默认0.9。

hive.mapjoin.followby.gby.localtask.max.memory.usage

如果 mapjoin 后面紧跟着一个 group by 任务，这种情况下本地任务的最大内存使用率，默认是0.55。

hive.mapjoin.check.memory.rows

localtask 每处理完多少行，就执行内存检查。默认为 100000。

Mapjoin 的局限性

下面是 mapjoin 的一些局限：

• 不能把全连接（full outer join）转换成 mapjoin。
• 可以把左连接转换成 mapjoin，但是右边的表的大小必须小于 25MB。
• 可以把右连接转换成 mapjoin，但是左边的表的大小必须小于 25MB。

如何知道 Hive 使用了 mapjoin

可以使用 explain 命令查看 HiveQL 代码的执行计划，在 Map 操作树我们可以看到 Map Side Join Operator 关键字，就说明 join 是 mapjoin。

hive> explain select a.* from passwords a,passwords3 b where a.col0=b.col0;
OK
STAGE DEPENDENCIES:
 Stage-4 is a root stage
 Stage-3 depends on stages: Stage-4
 Stage-0 is a root stage
STAGE PLANS:
 Stage: Stage-4
   Map Reduce Local Work
     Alias -> Map Local Tables:
       b
         Fetch Operator
           limit: -1
     Alias -> Map Local Operator Tree:
       b
         TableScan
           alias: b
           Statistics: Num rows: 1 Data size: 31 Basic stats: COMPLETE Column stats: NONE
           HashTable Sink Operator
             condition expressions:
               0 {col0} {col1} {col2} {col3} {col4} {col5} {col6}
               1 {col0}
             keys:
               0 col0 (type: string)
               1 col0 (type: string)
 Stage: Stage-3
   Map Reduce
     Map Operator Tree:
         TableScan
           alias: a
           Statistics: Num rows: 9963904 Data size: 477218560 Basic stats: COMPLETE Column stats: NONE
           Map Join Operator
             condition map:
                  Inner Join 0 to 1
             condition expressions:
               0 {col0} {col1} {col2} {col3} {col4} {col5} {col6}
               1 {col0}
keys:
               0 col0 (type: string)
               1 col0 (type: string)
             outputColumnNames: _col0, _col1, _col2, _col3, _col4, _col5, _col6, _col9
             Statistics: Num rows: 10960295 Data size: 524940416 Basic stats: COMPLETE Column stats: NONE
             Filter Operator
               predicate: (_col0 = _col9) (type: boolean)
               Statistics: Num rows: 5480147 Data size: 262470184 Basic stats: COMPLETE Column stats: NONE
               Select Operator
                 expressions: _col0 (type: string), _col1 (type: string), _col2 (type: string), _col3 (type: string), _col4 (type: string), _col5 (type: string), _col6 (type: string)
                 outputColumnNames: _col0, _col1, _col2, _col3, _col4, _col5, _col6
                 Statistics: Num rows: 5480147 Data size: 262470184 Basic stats: COMPLETE Column stats: NONE
                 File Output Operator
                   compressed: false
                   Statistics: Num rows: 5480147 Data size: 262470184 Basic stats: COMPLETE Column stats: NONE
                   table:
                       input format: org.apache.hadoop.mapred.TextInputFormat
                       output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
                       serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
     Local Work:
       Map Reduce Local Work
 Stage: Stage-0
   Fetch Operator
     limit: -1

使用 MAPJOIN 提示指定小表

可以使用 /*+ MAPJOIN(小表表名)*/ 提示让 Hive 执行引擎知道哪个表作为小表。代码如下：

hive> set hive.auto.convert.join=true;
hive> set hive.auto.convert.join.noconditionaltask=true;
hive> set hive.auto.convert.join.noconditionaltask.size=20971520
hive> set hive.auto.convert.join.use.nonstaged=true;
hive> set hive.mapjoin.smalltable.filesize = 30000000;
hive> set hive.ignore.mapjoin.hint=false;
hive> Select /*+ MAPJOIN(b) */ a.key, a.value 
from a join b
on a.key = b.key

上面代码指定 b 表作为小表。需要注意的是，使用 mapjoin 提示需要把 hive.ignore.mapjoin.hint 参数设置为 false，表示不要忽略 HQL 代码中的提示，该配置默认为 true。