你真的了解Lateral View explode吗？--源码复盘

数据仓库践行者

发布于 2020-05-07 14:59:33

2.2K0

发布于 2020-05-07 14:59:33

文章被收录于专栏：数据仓库践行者

用Lateral view explode这么久，竟然发现，不是很了解它？

Lateral view与UDTF函数一起使用，UDTF对每个输入行产生0或者多个输出行。Lateral view首先在基表的每个输入行应用UDTF，然后连接结果输出行与输入行组成拥有指定表别名的虚拟表。

UDTF函数需要继承GenericUDTF.java，在hive源码中，可以查到有以下8种UDTF函数：

我们最常用的就是explode了，可是如果面试的时候，我问你：Lateral view explode 会产生shuffle吗？为什么会，或者不会？

你确定你能毫不犹豫、确定无疑的答出来吗？

如果不能，那你真的需要看看这篇原理复盘的文章。

源码分析

该从哪里说起呢？写一个简单的sql：

explain SELECT id, sq,myCol from window_test_table LATERAL VIEW explode(split(sq,',')) myTab as myCol;

这个sql不多说，一眼看透，下面看一下执行计划：

上面我把执行计划的执行顺序给标注了一下，说实话，真的不太喜欢hive的执行计划，太长了，相反 spark的就很简洁，看一眼就知道怎么回事，但很多时候，还是喜欢来分解hive的实现过程，哈哈，这是不是找虐型的~~

看了我的标注后，一目了然，这个sql 经历了两条线：

ts(TableScan)-->lvf(Lateral View Forward)-->sel(Select)-->lvj(Lateral View Join)-->sel(Select)ts(TableScan)-->lvf(Lateral View Forward)-->sel(Select)-->udtf-->lvj(Lateral View Join)-->sel(Select)

1、TableScanOperator

不多说，常规读表操作

2、LateralViewForwardOperator

@Overridepublic void process(Object row, int tag) throws HiveException {  forward(row, inputObjInspectors[tag]);}

几乎什么都没做，数据怎么来的，还怎么送出去。

它的作用就是告知一下：

2-1、左侧SelectOperator

筛选出你需要的非explode的列：id,sq

2-2-1、右侧SelectOperator

筛选出explode的列：split(sq, ',')

2-2-2、右侧UDTFOperator

这个稍显复杂，代码里暗藏玄机

@Overridepublic void process(Object row, int tag) throws HiveException {  StructObjectInspector soi = (StructObjectInspector) inputObjInspectors[tag];  List<? extends StructField> fields = soi.getAllStructFieldRefs();  //从row里解出字段  for (int i = 0; i < fields.size(); i++) {    objToSendToUDTF[i] = soi.getStructFieldData(row, fields.get(i));  }

//真正处理数据的是 genericUDTF的某个实现类，比如，explode，那就是GenericUDTFExplode.java 的process  genericUDTF.process(objToSendToUDTF);  //这里判断一下有没有outer关键字。这里真的真的真的是，可能用了很久了，还不知道udtf还有个outer 关键字  if (conf.isOuterLV() && collector.getCounter() == 0) {    //思考一下这一步是干嘛？    collector.collect(outerObj);  }  collector.reset();}

GenericUDTFExplode.java就相当容易理解了，毕竟我们自己写udtf时，也是这么做的：

/** * GenericUDTFExplode. * */@Description(name = "explode",    value = "_FUNC_(a) - separates the elements of array a into multiple rows,"      + " or the elements of a map into multiple rows and columns ")public class GenericUDTFExplode extends GenericUDTF {....@Override//主要处理数据的方法public void process(Object[] o) throws HiveException {  switch (inputOI.getCategory()) {  case LIST:  //处理list    ListObjectInspector listOI = (ListObjectInspector)inputOI;    List<?> list = listOI.getList(o[0]);    if (list == null) {      return;  //当数组里没有值时，不发送数据    }    for (Object r : list) {      forwardListObj[0] = r;      forward(forwardListObj);    }    break;  case MAP: //处理map    MapObjectInspector mapOI = (MapObjectInspector)inputOI;    Map<?,?> map = mapOI.getMap(o[0]);    if (map == null) {      return;    }    for (Entry<?,?> r : map.entrySet()) {      forwardMapObj[0] = r.getKey();      forwardMapObj[1] = r.getValue();      forward(forwardMapObj);    }    break;  default:    throw new TaskExecutionException("explode() can only operate on an array or a map");  }}....}

上面所有的，都没有什么特别的，如果必须让找一个的话，那我选择outer关键字吧。

为什么要有outer关键字？

当UDTF不产生任何行时，比如explode()函数的输入列为空，LATERALVIEW就不会生成任何输出行。在这种情况下原有行永远不会出现在结果中。OUTRE可被用于阻止这种情况，输出行中来自UDTF的列将被设置为NULL。

看下图结果便一目了然：

实际上从代码里，也能够看到：

当没有值时，是return掉，不会forward，如果不forward的话，那这条数据就不会被传入下个Operator，也就不会被输出

那outer是怎么处理的呢？

UDTF会借助UDTFCollector为其展开的结果计数，并forward:

@Overridepublic void collect(Object input) throws HiveException {  op.forwardUDTFOutput(input);  counter++;}

如果没有展开结果，counter就为0。这样，进入outer之后，会把之前建好的没有内容的outerObj给forward到下个算子LateralViewJoinOperator

3、LateralViewJoinOperator

@Overridepublic void process(Object row, int tag) throws HiveException {  StructObjectInspector soi = (StructObjectInspector) inputObjInspectors[tag];//标识是左侧select过来的  if (tag == SELECT_TAG) {    selectObjs.clear();    selectObjs.addAll(soi.getStructFieldsDataAsList(row));  } else if (tag == UDTF_TAG) {  //代表是右侧udtf过来的    acc.clear();    acc.addAll(selectObjs);    acc.addAll(soi.getStructFieldsDataAsList(row));  //合并数据    forward(acc, outputObjInspector);  } else {    throw new HiveException("Invalid tag");  }

}

LateralViewJoinOperator处理逻辑也是很简单明了，这里的join也是简单的List.addAll

最后最后

那问题来了？

Lateral view explode 会产生shuffle吗？

当然不会，毋庸置疑！其实一开始看执行计划就会发现，没有reduce任务呀~~

这里的Join代表的是两份数据联接到一起的意思，并不是真正的意义上的join。

另外，还有一点：

大家要注意加outer和不加outer的区别，在日常工作一定要注意，切记切记！

不要丢了数据，还不知道为啥

最后最后最后

再考虑一下，在我们看来 Lateral view explode 仅仅是个很简单的数组 or map 结构展开，再联接的操作，几行代码就能搞定的事，为何，hive要设计的这么麻烦呢？毋庸置疑！-- 为了解耦。

UDTF 可以单独用，可以和Lateral view一起用，并且用户还可以定制自己的UDTF~~