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

Posted 小萝卜算子

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了你真的了解Lateral View explode吗?--源码复盘相关的知识,希望对你有一定的参考价值。

​这几天情绪真的很down啊,包括写这篇文章的时候,无论如何,希望大家看了这篇文章后,功力能再上涨一层~~

用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

@Override
public 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

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

@Override
public 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:

@Override
public void collect(Object input) throws HiveException 
  op.forwardUDTFOutput(input);
  counter++;

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

3、LateralViewJoinOperator

@Override
public 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~~  

这一切 都与hive这样灵活的设计分不开的

推荐阅读:

sparksql源码系列 | 生成resolved logical plan的解析规则整理

数仓面试——连续登录问题

数仓面试——连续登录问题进阶版

Spark sql 谓词下推之CombineFilters及可以改进的地方

以上是关于你真的了解Lateral View explode吗?--源码复盘的主要内容,如果未能解决你的问题,请参考以下文章

LATERAL VIEW EXPLODE与LATERAL VIEW OUTER EXPLODE的区别

explode和lateral view用法

Hive中grouping sets与lateral view explode的用法

Hive中grouping sets与lateral view explode的用法

Hive之explode()函数和posexplode()函数和lateral view函数

Hive行转列(explode/lateral view)