[学习交流] 【上海校区】join的工作原理和UDF的定义

深入join的过程
虽然我们在Spark篇的SparkSQL入门中也提到了一些join的东西，但是还是不够详细。这里我们将在Hive中执行join操作，看看在执行计划中究竟是怎么样的。
现在我们有这么一个SQL语句，里面是对两个表进行join：

select
a.empno,a.ename,a.deptno,b.dname
from
emp a join dept b
on a.deptno = b.deptno;

这里是执行这个操作的日志截图：

我们可以从这个日志上看到，我们将两个表进行了join操作却没有reduce操作，那么就是没有shuffle操作呗，这是不是很棒呀。但是我们现在需要把shuffle弄出来才方便我们理解下面的东西，所以我们设置一下这个参数：hive.auto.convert.join，现在这个值是true，我们将它设置成false在走一次看看情况是什么，然后再开始将。

通过上面的两次操作，我们知道原来join还可以有两种的，一种是没有reduce的一种是有reduce的。

一般正常的join我们其实可以把其定义成一中CommonJoin/ReduceJoin，这种join就是在reduce端执行的，所以过程肯定是伴随着shuffle的；而另外一种join就是没有shuffle的，这个join是发生在map端的，所以我们可以定义成为MapJoin(其实就是使用了类似Spark过程呃广播变量)

CommonJoin
我们先来介绍CommonJoin，同时结合执行计划来观察

explain
select
a.empno,a.ename,a.deptno,b.dname
from
emp a join dept b
on a.deptno = b.deptno;

得到的结果是：

STAGE DEPENDENCIES:
  Stage-1 is a root stage
  Stage-0 depends on stages: Stage-1

STAGE PLANS:
  Stage: Stage-1
Map Reduce
   Map Operator Tree:
      TableScan
         alias: a
         Statistics: Num rows: 6 Data size: 701 Basic stats: COMPLETE Column stats: NONE
         Filter Operator
            predicate: deptno is not null (type: boolean)
            Statistics: Num rows: 3 Data size: 350 Basic stats: COMPLETE Column stats: NONE
            Reduce Output Operator
            key expressions: deptno (type: int)
            sort order: +
            Map-reduce partition columns: deptno (type: int)
            Statistics: Num rows: 3 Data size: 350 Basic stats: COMPLETE Column stats: NONE
            value expressions: empno (type: int), ename (type: string)
      TableScan
         alias: b
         Statistics: Num rows: 1 Data size: 80 Basic stats: COMPLETE Column stats: NONE
         Filter Operator
            predicate: deptno is not null (type: boolean)
            Statistics: Num rows: 1 Data size: 80 Basic stats: COMPLETE Column stats: NONE
            Reduce Output Operator
            key expressions: deptno (type: int)
            sort order: +
            Map-reduce partition columns: deptno (type: int)
            Statistics: Num rows: 1 Data size: 80 Basic stats: COMPLETE Column stats: NONE
            value expressions: dname (type: string)
   Reduce Operator Tree:
      Join Operator
      condition map:
            Inner Join 0 to 1
      keys:
         0 deptno (type: int)
         1 deptno (type: int)
      outputColumnNames: _col0, _col1, _col7, _col12
      Statistics: Num rows: 3 Data size: 385 Basic stats: COMPLETE Column stats: NONE
      Select Operator
         expressions: _col0 (type: int), _col1 (type: string), _col7 (type: int), _col12 (type: string)
         outputColumnNames: _col0, _col1, _col2, _col3
         Statistics: Num rows: 3 Data size: 385 Basic stats: COMPLETE Column stats: NONE
         File Output Operator
            compressed: false
            Statistics: Num rows: 3 Data size: 385 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

  Stage: Stage-0
Fetch Operator
   limit: -1
   Processor Tree:
      ListSink

这里有两个stage
stage-1中先是一个map阶段，里面先扫描表a，然后执行一个filter阶段(这个个阶段只用你的SQL里有join都会有的)，并得到所需要的字段：a.empno,a.ename,a.deptno，其中a.deptno是key用来进行join操作的。然后即使扫描表b，同上的操作。
然后就是一个reduce阶段了，在这个阶段就会执行join操作了。
就是先对emp表进行一个map得到一堆mappers，同样的操作也对dept表执行了一次也是得到一堆的mappers，接着一个shuffle过程将相同的key的数据重新分配到相同的位置上，最后执行reduce操作。

MapJoin
我们修改hive.auto.convert.join的值为true，在来看看执行计划是怎么样：

STAGE DEPENDENCIES:
  Stage-4 is a root stage
  Stage-3 depends on stages: Stage-4
  Stage-0 depends on stages: Stage-3

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: 80 Basic stats: COMPLETE Column stats: NONE
         Filter Operator
            predicate: deptno is not null (type: boolean)
            Statistics: Num rows: 1 Data size: 80 Basic stats: COMPLETE Column stats: NONE
            HashTable Sink Operator
            keys:
               0 deptno (type: int)
               1 deptno (type: int)

  Stage: Stage-3
Map Reduce
   Map Operator Tree:
      TableScan
         alias: a
         Statistics: Num rows: 6 Data size: 701 Basic stats: COMPLETE Column stats: NONE
         Filter Operator
            predicate: deptno is not null (type: boolean)
            Statistics: Num rows: 3 Data size: 350 Basic stats: COMPLETE Column stats: NONE
            Map Join Operator
            condition map:
                  Inner Join 0 to 1
            keys:
               0 deptno (type: int)
               1 deptno (type: int)
            outputColumnNames: _col0, _col1, _col7, _col12
            Statistics: Num rows: 3 Data size: 385 Basic stats: COMPLETE Column stats: NONE
            Select Operator
               expressions: _col0 (type: int), _col1 (type: string), _col7 (type: int), _col12 (type: string)
               outputColumnNames: _col0, _col1, _col2, _col3
               Statistics: Num rows: 3 Data size: 385 Basic stats: COMPLETE Column stats: NONE
               File Output Operator
                  compressed: false
                  Statistics: Num rows: 3 Data size: 385 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
   Processor Tree:
      ListSink

这次stage就有3个了，首先执行一个Map Reduce Local Work工作。将b表转换成一个Map Local Table，就是广播到各个节点上去，再扫描b表定义好需要的key。就到下一个stage，在这个stage上面，就是扫描a表，定义好key，然后就执行了join操作了。最后就是一个stage-0，也没有了reduce操作。
具体就是，将一个小的表生成一个HashTable，然后加载到分布式缓存上面，接着为大表生成一堆的mappers，然后从分布式缓存里面将小表生成的HashTable和大表的每一个mappers进行join，得到我们需要的输出。
从运行日志上看：
2019-01-09 05:33:59 Starting to launch local task to process map join; maximum memory = 477626368
2019-01-09 05:34:00 Dump the side-table for tag: 1 with group count: 4 into file: file:/tmp/test/5180d098-6132-43ab-a32a-37688d2de617/hive_2019-01-09_17-33-56_429_8137188178107053233-1/-local-10003/HashTable-Stage-3/MapJoin-mapfile31–.hashtable
2019-01-09 05:34:00 Uploaded 1 File to: file:/tmp/test/5180d098-6132-43ab-a32a-37688d2de617/hive_2019-01-09_17-33-56_429_8137188178107053233-1/-local-10003/HashTable-Stage-3/MapJoin-mapfile31–.hashtable (373 bytes)
2019-01-09 05:34:00 End of local task; Time Taken: 1.22 sec.
Execution completed successfully
Launching Job 1 out of 1
Number of reduce tasks is set to 0 since there’s no reduce operator
Starting Job = job_1547022064410_0004, Tracking URL = http://doudou:8288/proxy/application_1547022064410_0004/
Kill Command = /home/test/app/hadoop/bin/hadoop job -kill job_1547022064410_0004
Hadoop job information for Stage-3: number of mappers: 1; number of reducers: 0
2019-01-09 17:34:07,909 Stage-3 map = 0%, reduce = 0%
2019-01-09 17:34:13,193 Stage-3 map = 100%, reduce = 0%, Cumulative CPU 1.91 sec
MapReduce Total cumulative CPU time: 1 seconds 910 msec
Ended Job = job_1547022064410_0004
MapReduce Jobs Launched:
Stage-Stage-3: Map: 1 Cumulative CPU: 1.91 sec HDFS Read: 6665 HDFS Write: 310 SUCCESS
Total MapReduce CPU Time Spent: 1 seconds 910 mse
先执行了一个local task(先定义好一个小表，这个的小表是小于一个具体的内存值，这个值可以我们自己手动设置的)，然后Dump the side-table for tag(就是生辰一个hashtable)，接着Uploaded 1 File to(这里这个file就是hashtable，看第三条日志)，最后完成local task后才开始执行map操作，没有reduce操作。

UDF注册
一般我们需要注册一个UDF的话需要现在idea上以java代码写好我们自定义的函数的实体，所以我们需要一个依赖：

org.apache.hive
hive-exec
${hive.version}

代码主体

//类名可以自定义，但一定要继承UDF这个类
public class doudou extends UDF{
  //方法名一定要是evaluate，但是返回值类型和参数列表可以自定义，但是因为是UDF所以返回值不要是集合类型的
  public String evaluate(String str){

return "xxx" + str;

  }
}

然后打成jar报，因为我们在Hive上面使用这个方法需要到这个jar包

注册UDF方法
注册UDF方法有三种方法，前两中是临时版的，当前会话没了就没了
方法一：
先找个文件加=夹放我们的jar包，比如：/home/doudou/app/hive/lib/doudou-udf.jar(jar包的全路径)
在Hive窗口里面：

add jar /home/doudou/app/hive/lib/doudou-udf.jar
create tempoary function douUDF as 'udfCreate.doudou';

这里as后面跟的是你的包名加类名
方法二：
如果不想输入这么一句话：add jar /home/doudou/app/hive/lib/doudou-udf.jar
可以在hive目录下创建一个目录：auxlib(一定要是这个目录名哦，想改的话就要修改配置文件里面的参数了)，把你的jar包放到里面去，那么每一次就是输入：create tempoary function douUDF as ‘udfCreate.doudou’;这句就好了。
方法三：
不想每一次都创建的话，可以使用下面这个方法，将这个函数注册到metastore里面

create function [db_name.]function_name as 'package_name.class_name' using jar 'hdfs://ip:port/some/path...'

这个方法就是将你的jar包放到HDFS上面，然后注册的时候使用using jar '…'来指定HDFS上的路径，然后你就能在你的MySQL上面的元数据地址看到这个函数了，可以直接使用了，只是每次打开一个会话时第一次使用这个函数都会自动加载一下的。如果哪位小伙伴不想这样的话，可以修改源码把方法写到源码上面去，然后重新编译一下，或者替换一下你现在使用的Hive目录里面lib目录下的hive-exec-1.1.0-cdh5.7.0.jar这个表，就是我们用idea编写我们自定义方法体的那个依赖，在里面添加你的函数。

不二晨 · 不二晨

奈斯，感谢分享

帐号		自动登录	找回密码
密码			加入黑马

[学习交流] 【上海校区】join的工作原理和UDF的定义

1 个回复