hive谓词下放

作者: 乌拉乌拉儿 | 来源:发表于2020-01-04 11:51 被阅读0次

保留表(Preserved Row table)

在outer join中需要返回所有数据的表叫做保留表，也就是说在left outer join中，左表需要返回所有数据，则左表是保留表；right outer join中右表则是保留表；在full outer join中左表和右表都要返回所有数据，则左右表都是保留表。

Null Supplying table

在outer join中对于没有匹配到的行需要用null来填充的表称为Null Supplying table。在left outer join中，左表的数据全返回，对于左表在右表中无法匹配的数据的相应列用null表示，则此时右表是Null Supplying table，相应的如果是right outer join的话，左表是Null Supplying table。但是在full outer join中左表和右表都是Null Supplying table，因为左表和右表都会用null来填充无法匹配的数据。

此时你会发现full outer join就是个矛盾体，因为此时的左表和右表即使保留表又是Null Supplying table，这就导致在outer join中谓词下放时有一些问题。

Join中的谓词

Join中的谓词是指 Join On语句中的谓词。如：’R1 join R2 on R1.x = 5’ the predicate ‘R1.x = 5’是Join中的谓词

Join之后的谓词

where语句中的谓词称之为Join之后的谓词

谓词下放

谓词：谓词是一个属性或是一个表示“持有”或“不持有”的表达式，换句话说，也就是取值为 TRUE、FALSE 或 UNKNOWN 的表达式。谓词用于 WHERE 子句和 HAVING 子句的搜索条件中，还用于 FROM 子句的联接条件以及需要布尔值的其他构造中。

对于outer join中的坑，主要是对谓词下放的规则不熟悉，导致理解的不够全面，造成一些坑。
谓词下放的规则有两个，分别为

1.Join中谓词如果是保留表的，则不会下放。
2.Join之后的谓词如果是Null Supplying tables的，则不会下放。

栗子

创建表

create table t1 (id int, name string) ROW FORMAT DELIMITED FIELDS TERMINATED BY “,”;
create table t2 (id int, name string) ROW FORMAT DELIMITED FIELDS TERMINATED BY “,”;

t1里的数据如下：

1,a
1,b
2,a
2,b
3,a
3,b
4,b

t2里的数据如下：

1,aa
4,dd

在执行例子之前，避免map join影响到各个例子的执行计划，先关闭map join，set hive.auto.convert.join=false;

demo1:

select t1.*,t2.name as name_t2 
from t1 
left outer join t2 
on (t1.id=t2.id) 
where t1.name='a' and t2.name='aa';

此sql中，保留表为t1，Null Supplying table为t2，join中谓词在此sql中没有，where中的两个谓词都是join之后的谓词。
则根据谓词下放规则，t1.name='a'是保留表的字段，则被下放，而t2.name='aa'是Null Supplying table的，Null Supplying table不支持join之后的谓词下放。
那么此时的sql就等同于

select t.*,t2.name as name_t2 
 from 
 (
    select id, name from t1 where t1.name='a'
 ) as t left outer join t2 on
 (t.id=t2.id) where t2.name='aa';

其结果为:

1 a aa

造成这样的结果是因为hive中outer join谓词下放的规则,sql中的t2.name='aa'是join之后的谓词，不会被下放到t2中(t2是Null Supplying table)，而是对t1和t2 join的结果进行filter，filter的条件是t2.name='aa'，于是就出现了1 a aa结果，因为其余数据的t2.name的值都为NULL</p>
首先sql中有两个join之后的谓词，分别为t1.name='a' and t2.name='aa'，其中t1.name是保留表t1的，会谓词下放，在scan t1表时，对t1中的数据进行过滤，t2.name是t2的，t2不是保留表，谓词无法下放，而where又是在join之后执行的(Joins occur BEFORE WHERE CLAUSES)。看下sql的执行计划：</p>

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: t1
              Statistics: Num rows: 5 Data size: 28 Basic stats: COMPLETE Column stats: NONE
              Filter Operator
                predicate: (name = 'a') (type: boolean)
                Statistics: Num rows: 2 Data size: 11 Basic stats: COMPLETE Column stats: NONE
                Reduce Output Operator
                  key expressions: id (type: int)
                  sort order: +
                  Map-reduce partition columns: id (type: int)
                  Statistics: Num rows: 2 Data size: 11 Basic stats: COMPLETE Column stats: NONE
            TableScan
              alias: t2
              Statistics: Num rows: 1 Data size: 10 Basic stats: COMPLETE Column stats: NONE
              Reduce Output Operator
                key expressions: id (type: int)
                sort order: +
                Map-reduce partition columns: id (type: int)
                Statistics: Num rows: 1 Data size: 10 Basic stats: COMPLETE Column stats: NONE
                value expressions: name (type: string)
        Reduce Operator Tree:
          Join Operator
            condition map:
                 Left Outer Join0 to 1
            keys:
              0 id (type: int)
              1 id (type: int)
            outputColumnNames: _col0, _col6
            Statistics: Num rows: 2 Data size: 12 Basic stats: COMPLETE Column stats: NONE
            Filter Operator
              predicate: (_col6 = 'aa') (type: boolean)
              Statistics: Num rows: 1 Data size: 6 Basic stats: COMPLETE Column stats: NONE
              Select Operator
                expressions: _col0 (type: int), 'a' (type: string), 'aa' (type: string)
                outputColumnNames: _col0, _col1, _col2
                Statistics: Num rows: 1 Data size: 6 Basic stats: COMPLETE Column stats: NONE
                File Output Operator
                  compressed: false
                  Statistics: Num rows: 1 Data size: 6 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

从sql的执行计划上也可以看出，TableScan t1时使用了predicate: (name = 'a') (type: boolean)对数据进行过滤，(则进入mr的t1数据不是全表数据，而是过滤过的数据)。而where对应的filter operator是在join之后才执行的
demo2:

select t1.*,t2.name as name_t2 
from t1 
left outer join t2 
on (t1.id=t2.id and t2.name='aa') w
here t1.name='a';

执行结果为:

 1  a   aa
 2  a   NULL
 3  a   NULL

其中t2.name是join中谓词，t1.name是join之后谓词，根据规则join中谓词能下放到t2中，join之后的谓词能下放到t1中，则上面sql的意思是将t1中name为a的数据与t2中name为aa的数据进行join，t1和t2都会在table scan时进行数据过滤。sql的执行计划如下：

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: t1
             Statistics: Num rows: 5 Data size: 28 Basic stats: COMPLETE Column stats: NONE
             Filter Operator
               predicate: (name = 'a') (type: boolean)
               Statistics: Num rows: 2 Data size: 11 Basic stats: COMPLETE Column stats: NONE
               Reduce Output Operator
                 key expressions: id (type: int)
                 sort order: +
                 Map-reduce partition columns: id (type: int)
                 Statistics: Num rows: 2 Data size: 11 Basic stats: COMPLETE Column stats: NONE
           TableScan
             alias: t2
             Statistics: Num rows: 1 Data size: 10 Basic stats: COMPLETE Column stats: NONE
             Filter Operator
               predicate: (name = 'aa') (type: boolean)
               Statistics: Num rows: 1 Data size: 10 Basic stats: COMPLETE Column stats: NONE
               Reduce Output Operator
                 key expressions: id (type: int)
                 sort order: +
                 Map-reduce partition columns: id (type: int)
                 Statistics: Num rows: 1 Data size: 10 Basic stats: COMPLETE Column stats: NONE
                 value expressions: name (type: string)
       Reduce Operator Tree:
         Join Operator
           condition map:
                Left Outer Join0 to 1
           keys:
             0 id (type: int)
             1 id (type: int)
           outputColumnNames: _col0, _col6
           Statistics: Num rows: 2 Data size: 12 Basic stats: COMPLETE Column stats: NONE
           Select Operator
             expressions: _col0 (type: int), 'a' (type: string), _col6 (type: string)
             outputColumnNames: _col0, _col1, _col2
             Statistics: Num rows: 2 Data size: 12 Basic stats: COMPLETE Column stats: NONE
             File Output Operator
               compressed: false
               Statistics: Num rows: 2 Data size: 12 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

从执行计划上可以看出t1和t2的TableScan中都存在Filter Operator操作，对数据进行过滤。
demo3:

select t1.*,t2.name as name_t2 from t1 left outer join t2 on (t1.id=t2.id and t1.name='a' ) where t2.name='aa';

此sql中有两个谓词，一个是join中谓词t1.name='a'和join之后谓词t2.name='aa'。
这里需要明白一点，join中谓词决定了进行join的数据，join之后谓词决定了最终要呈现的数据。也就是说join中谓词t1.name='a'决定了t1表中name为a的数据才去和t2进行join，而join之后谓词t2.name='aa'决定了最终的结果。
执行结果为：

1 a aa

这里的结果随便和case 1的结果一样，但是执行的内部逻辑是不一样的，case3在执行where filter之前返回的数据是t1全表的数据，而case1在执行where filter之前返回的数据是t1中name为a的数据，因为case3中join中谓词t1.name='a'并不能对t1进行下放，这里的语义是对t1进行全表扫描，只拿name为a的数据和t2进行join，id匹配成功则为t2.name的值，匹配不成功和name不为a的数据中t2.name的值都为NULL，join结束之后由where filter过滤t2.name='aa'，因此就出现了上面的结果(因为没有匹配成功的和没有进行匹配的数据都是NULL，被过滤)。

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: t1
             Statistics: Num rows: 1 Data size: 28 Basic stats: COMPLETE Column stats: NONE
             Reduce Output Operator
               key expressions: id (type: int)
               sort order: +
               Map-reduce partition columns: id (type: int)
               Statistics: Num rows: 1 Data size: 28 Basic stats: COMPLETE Column stats: NONE
               value expressions: name (type: string)
           TableScan
             alias: t2
             Statistics: Num rows: 1 Data size: 10 Basic stats: COMPLETE Column stats: NONE
             Reduce Output Operator
               key expressions: id (type: int)
               sort order: +
               Map-reduce partition columns: id (type: int)
               Statistics: Num rows: 1 Data size: 10 Basic stats: COMPLETE Column stats: NONE
               value expressions: name (type: string)
       Reduce Operator Tree:
         Join Operator
           condition map:
                Left Outer Join0 to 1
           filter predicates:
             0 {(VALUE._col0 = 'a')}
             1
           keys:
             0 id (type: int)
             1 id (type: int)
           outputColumnNames: _col0, _col1, _col6
           Statistics: Num rows: 1 Data size: 30 Basic stats: COMPLETE Column stats: NONE
           Filter Operator
             predicate: (_col6 = 'aa') (type: boolean)
             Statistics: Num rows: 1 Data size: 30 Basic stats: COMPLETE Column stats: NONE
             Select Operator
               expressions: _col0 (type: int), _col1 (type: string), 'aa' (type: string)
               outputColumnNames: _col0, _col1, _col2
               Statistics: Num rows: 1 Data size: 30 Basic stats: COMPLETE Column stats: NONE
               File Output Operator
                 compressed: false
                 Statistics: Num rows: 1 Data size: 30 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

上面是demo3的执行计划，可以看出两个表在TableScan时都是全表扫描，没有进行filter，在join时有个filterfilter predicates，join结束之后又有个Filter Operator对join的结果进行过滤。
demo4:

select t1.*,t2.name as name_t2 from t1 left outer join t2 on (t1.id=t2.id and t1.name='a' and t2.name='aa');

此sql中只有join中谓词，根据规则t1.name=’a’不能下放到t1中，而t2.name=’aa’能够下放到t2中，那就是说扫t1的全表数据，然后拿t1.name为a的数据去和t2中t2.name为aa的数据(t2在table scan时会过滤name为aa的数据)进行join，那么t1中name不为a的数据不和t2进行join，直接将t2.name置为NULL，而t1中name为a却没有和t2进行join成功的数据也为NULL，所以其结果如下：

 1  b   NULL
 1  a   aa
 2  b   NULL
 2  a   NULL
 3  b   NULL
 3  a   NULL
 4  b   NULL

这里需要说明下，t1是保留表，join之后要返回其表中所有的数据，因为在join之后没有where对其结果进行过滤，所以此处显示的是t1表的所有数据，上面的case之所以没有返回t1的所有数据是因为他们要么在table scan时进行了过滤要么就是在join之后通过where进行了过滤

为了验证下t1中只有那么为a的数据与t2进行了join，我对t2中的数据进行了修改，添加一条数据4,aa，其执行结果按照上面的逻辑应该不会发生变化，再次执行case4，看下返回的结果是否一致。其结果如下:

  1 b   NULL
  1 a   aa
  2 b   NULL
  2 a   NULL
  3 b   NULL
  3 a   NULL
  4 b   NULL

与上面的结果一致。
此sql的执行计划如下:

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: t1
             Statistics: Num rows: 1 Data size: 28 Basic stats: COMPLETE Column stats: NONE
             Reduce Output Operator
               key expressions: id (type: int)
               sort order: +
               Map-reduce partition columns: id (type: int)
               Statistics: Num rows: 1 Data size: 28 Basic stats: COMPLETE Column stats: NONE
               value expressions: name (type: string)
           TableScan
             alias: t2
             Statistics: Num rows: 1 Data size: 10 Basic stats: COMPLETE Column stats: NONE
             Filter Operator
               predicate: (name = 'aa') (type: boolean)
               Statistics: Num rows: 1 Data size: 10 Basic stats: COMPLETE Column stats: NONE
               Reduce Output Operator
                 key expressions: id (type: int)
                 sort order: +
                 Map-reduce partition columns: id (type: int)
                 Statistics: Num rows: 1 Data size: 10 Basic stats: COMPLETE Column stats: NONE
                 value expressions: name (type: string)
       Reduce Operator Tree:
         Join Operator
           condition map:
                Left Outer Join0 to 1
           filter predicates:
             0 {(VALUE._col0 = 'a')}
             1
           keys:
             0 id (type: int)
             1 id (type: int)
           outputColumnNames: _col0, _col1, _col6
           Statistics: Num rows: 1 Data size: 30 Basic stats: COMPLETE Column stats: NONE
           Select Operator
             expressions: _col0 (type: int), _col1 (type: string), _col6 (type: string)
             outputColumnNames: _col0, _col1, _col2
             Statistics: Num rows: 1 Data size: 30 Basic stats: COMPLETE Column stats: NONE
             File Output Operator
               compressed: false
               Statistics: Num rows: 1 Data size: 30 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

这四个case介绍了left outer join的情况，right outer join与left类似，只是保留表和Null Supplying table换下位置，比较特殊的是full outer join，在full outer join中join中谓词和join之后谓词都不会被下放，(这里说不会下放可能不太准备，会下放，但是下放的位置不对，下放前和下放后的位置是一样的，开启log可以查看。)

因为full outer join中的两个表既是保留表也是Null Supplying table。来看个full outer join的例子，

 select t1.*,t2.name as name_t2 from t1 full outer join t2 on (t1.id=t2.id and t2.name='aa') where t1.name='a';

其结果为

  1 a   aa
  2 a   NULL
  3 a   NULL

其sql的执行计划如下：

  STAGE PLANS:
     Stage: Stage-1
       Map Reduce
         Map Operator Tree:
             TableScan
               alias: t1
               Statistics: Num rows: 1 Data size: 28 Basic stats: COMPLETE Column stats: NONE
               Reduce Output Operator
                 key expressions: id (type: int)
                 sort order: +
                 Map-reduce partition columns: id (type: int)
                 Statistics: Num rows: 1 Data size: 28 Basic stats: COMPLETE Column stats: NONE
                 value expressions: name (type: string)
             TableScan
               alias: t2
               Statistics: Num rows: 1 Data size: 15 Basic stats: COMPLETE Column stats: NONE
               Reduce Output Operator
                 key expressions: id (type: int)
                 sort order: +
                 Map-reduce partition columns: id (type: int)
                 Statistics: Num rows: 1 Data size: 15 Basic stats: COMPLETE Column stats: NONE
                 value expressions: name (type: string)
         Reduce Operator Tree:
           Join Operator
             condition map:
                  Outer Join 0 to 1
             filter predicates:
               0
               1 {(VALUE._col0 = 'aa')}
             keys:
               0 id (type: int)
               1 id (type: int)
             outputColumnNames: _col0, _col1, _col6
             Statistics: Num rows: 1 Data size: 30 Basic stats: COMPLETE Column stats: NONE
             Filter Operator
               predicate: (_col1 = 'a') (type: boolean)
               Statistics: Num rows: 1 Data size: 30 Basic stats: COMPLETE Column stats: NONE
               Select Operator
                 expressions: _col0 (type: int), 'a' (type: string), _col6 (type: string)
                 outputColumnNames: _col0, _col1, _col2
                 Statistics: Num rows: 1 Data size: 30 Basic stats: COMPLETE Column stats: NONE
                 File Output Operator
                   compressed: false
                   Statistics: Num rows: 1 Data size: 30 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

从执行计划中可以看出t1和t2在scan时都没有进行filter过滤，也就是说join中谓词和join之后的谓词都没有进行下放，join中谓词是在Join Operator时通过filter predicates进行过滤join，而join之后谓词是在join之后通过Filter Operator进行过滤的。

如果t1和t2某个表是分区表，此时想只full outer join某个分区的数据，应该怎么过滤更高效呢？
此时应该使用子查询，sql为select t1.*,t2.name as name_t2 from t1 full outer join t2 on (t1.id=t2.id and t2.name='aa')

select t.*,t2.name as name_t2 
  from
     （
       select * 
         from t1 
        where day='20170411'
      ) t
  full outer join  t2
    on t.id=t2.id

转载：hive outer join 之谓词下放

hive谓词下放

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

大数据那些事