博客

Hive – Group By 的实现 explain分析

  • Post author:
  • Post category:其他




准备数据


[java]


view plain


copy



  1. SELECT uid, SUM(COUNT) FROM logs GROUP BY uid;


[java]


view plain


copy



  1. hive> SELECT * FROM logs;

  2. a   苹果

    5

  3. a   橙子

    3

  4. a      苹果

    2

  5. b   烧鸡

    1


  7. hive> SELECT uid, SUM(COUNT) FROM logs GROUP BY uid;

  8. a

    10

  9. b

    1



计算过程


hive-groupby-cal


默认设置了hive.map.aggr=true，所以会在mapper端先group by一次，最后再把结果merge起来，为了减少reducer处理的数据量。注意看explain的mode是不一样的。mapper是hash，reducer是mergepartial。如果把hive.map.aggr=false，那将groupby放到reducer才做，他的mode是complete.



Operator


hive-groupby-op



Explain


[java]


view plain


copy



  1. hive> explain SELECT uid, sum(count) FROM logs group by uid;

  2. OK

  3. ABSTRACT SYNTAX TREE:

  4. (TOK_QUERY (TOK_FROM (TOK_TABREF (TOK_TABNAME logs))) (TOK_INSERT (TOK_DESTINATION (TOK_DIR TOK_TMP_FILE)) (TOK_SELECT (TOK_SELEXPR (TOK_TABLE_OR_COL uid)) (TOK_SELEXPR (TOK_FUNCTION sum (TOK_TABLE_OR_COL count)))) (TOK_GROUPBY (TOK_TABLE_OR_COL uid))))


  6. STAGE DEPENDENCIES:

  7. Stage-

    1


    is a root stage

  8. Stage-

    0


    is a root stage


  10. STAGE PLANS:

  11. Stage: Stage-

    1

  12. Map Reduce

  13. Alias -> Map Operator Tree:

  14. logs

  15. TableScan

    // 扫描表

  16. alias: logs

  17. Select Operator

    //选择字段

  18. expressions:

  19. expr: uid

  20. type: string

  21. expr: count

  22. type:

    int

  23. outputColumnNames: uid, count

  24. Group By Operator

    //这里是因为默认设置了hive.map.aggr=true，会在mapper先做一次聚合，减少reduce需要处理的数据

  25. aggregations:

  26. expr: sum(count)

    //聚集函数

  27. bucketGroup:

    false

  28. keys:

    //键

  29. expr: uid

  30. type: string

  31. mode: hash

    //hash方式，processHashAggr()

  32. outputColumnNames: _col0, _col1

  33. Reduce Output Operator

    //输出key，value给reducer

  34. key expressions:

  35. expr: _col0

  36. type: string

  37. sort order: +

  38. Map-reduce partition columns:

  39. expr: _col0

  40. type: string

  41. tag: –

    1

  42. value expressions:

  43. expr: _col1

  44. type: bigint

  45. Reduce Operator Tree:

  46. Group By Operator


  48. aggregations:

  49. expr: sum(VALUE._col0)


  50. //聚合

  51. bucketGroup:

    false

  52. keys:

  53. expr: KEY._col0

  54. type: string

  55. mode: mergepartial

    //合并值

  56. outputColumnNames: _col0, _col1

  57. Select Operator

    //选择字段

  58. expressions:

  59. expr: _col0

  60. type: string

  61. expr: _col1

  62. type: bigint

  63. outputColumnNames: _col0, _col1

  64. File Output Operator

    //输出到文件

  65. compressed:

    false

  66. GlobalTableId:

    0

  67. table:

  68. input format: org.apache.hadoop.mapred.TextInputFormat

  69. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  71. Stage: Stage-

    0

  72. Fetch Operator

  73. limit: –

    1


  75. hive> select distinct value from src;

  76. hive> select max(key) from src;


  78. 因为没有grouping keys，所以只有一个reducer。







  84. 2.2


    如果有聚合函数或者groupby，做如下处理：


  86. 插入一个select operator，选取所有的字段，用于优化阶段ColumnPruner的优化



  88. 2.2


    .


    1


    hive.map.aggr为


    true


    ，默认是


    true


    ，开启的，在map端做部分聚合



  90. 2.2


    .


    1.1


    hive.groupby.skewindata为


    false


    ，默认是关闭的，groupby的数据没有倾斜。


  92. 生成的operator是: GroupByOperator+ReduceSinkOperator+GroupByOperator。


  94. GroupByOperator+ReduceSinkOperator用于在map端做操作，第一个GroupByOperator在map端先做部分聚合。第二个用于在reduce端做GroupBy操作



  96. 2.2


    .


    1.2


    hive.groupby.skewindata为


    true


  98. 生成的operator是: GroupbyOperator+ReduceSinkOperator+GroupbyOperator+ReduceSinkOperator +GroupByOperator


  100. GroupbyOperator+ReduceSinkOperator（第一个MapredTask的map阶段）


  102. GroupbyOperator(第一个MapredTask的reduce阶段)


  104. ReduceSinkOperator （第二个MapredTask的map阶段）


  106. GroupByOperator（第二个MapredTask的reduce阶段）



  108. 2.2


    .


    2


    hive.map.aggr为


    false



  110. 2.2


    .


    2.1


    hive.groupby.skewindata为


    true


  112. 生成的operator是: ReduceSinkOperator+GroupbyOperator+ReduceSinkOperator +GroupByOperator


  114. ReduceSinkOperator(第一个MapredTask的map阶段)


  116. GroupbyOperator(第一个MapredTask的reduce阶段)


  118. ReduceSinkOperator （第二个MapredTask的map阶段）


  120. GroupByOperator（第二个MapredTask的reduce阶段）



  122. 2.2


    .


    2.2


    hive.groupby.skewindata为


    false


  124. 生成的operator是: ReduceSinkOperator(map阶段运行)+GroupbyOperator（reduce阶段运行）








  132. 第一种情况：


  134. set hive.map.aggr=

    false


    ;


  136. set hive.groupby.skewindata=

    false


    ;


  138. SemanticAnalyzer.genGroupByPlan1MR(){


  140. （

    1


    ）ReduceSinkOperator： It will put all Group By keys and the

  141. distinct field (

    if


    any) in the map-reduce sort key, and all other fields

  142. in the map-reduce value.


  144. （

    2


    ）GroupbyOperator：GroupByDesc.Mode.COMPLETE，Reducer: iterate/merge (mode = COMPLETE)


  146. }




  150. 第二种情况：


  152. set hive.map.aggr=

    true


    ;


  154. set hive.groupby.skewindata=

    false


    ;


  156. SemanticAnalyzer.genGroupByPlanMapAggr1MR(){


  158. （

    1


    ）GroupByOperator：GroupByDesc.Mode.HASH，The agggregation

  159. evaluation functions are as follows: Mapper: iterate/terminatePartial

  160. (mode = HASH)


  162. （

    2


    ）ReduceSinkOperator：Partitioning Key: grouping key。Sorting Key:

  163. grouping key

    if


    no DISTINCT grouping + distinct key


    if


    DISTINCT


  165. （

    3


    ）GroupByOperator：GroupByDesc.Mode.MERGEPARTIAL，Reducer:

  166. iterate/terminate

    if


    DISTINCT merge/terminate


    if


    NO DISTINCT (mode =

  167. MERGEPARTIAL)


  169. }




  173. 第三种情况：


  175. set hive.map.aggr=

    false


    ;


  177. set hive.groupby.skewindata=

    true


    ;


  179. SemanticAnalyzer.genGroupByPlan2MR(){


  181. （

    1


    ）ReduceSinkOperator：Partitioning Key: random()


    if


    no DISTINCT

  182. grouping + distinct key

    if


    DISTINCT。Sorting Key: grouping key


    if


    no

  183. DISTINCT grouping + distinct key

    if


    DISTINCT


  185. （

    2


    ）GroupbyOperator：GroupByDesc.Mode.PARTIAL1，Reducer: iterate/terminatePartial (mode = PARTIAL1)


  187. （

    3


    ）ReduceSinkOperator：Partitioning Key: grouping key。Sorting

  188. Key: grouping key

    if


    no DISTINCT grouping + distinct key


    if


    DISTINCT


  190. （

    4


    ）GroupByOperator：GroupByDesc.Mode.FINAL，Reducer: merge/terminate (mode = FINAL)


  192. }




  196. 第四种情况：


  198. set hive.map.aggr=

    true


    ;


  200. set hive.groupby.skewindata=

    true


    ;


  202. SemanticAnalyzer.genGroupByPlanMapAggr2MR(){


  204. （

    1


    ）GroupbyOperator：GroupByDesc.Mode.HASH，Mapper: iterate/terminatePartial (mode = HASH)


  206. （

    2


    ）ReduceSinkOperator： Partitioning Key: random()


    if


    no

  207. DISTINCT grouping + distinct key

    if


    DISTINCT。 Sorting Key: grouping key


  208. if


    no DISTINCT grouping + distinct key


    if


    DISTINCT。


  210. （

    3


    ）GroupbyOperator：GroupByDesc.Mode.PARTIALS， Reducer:

  211. iterate/terminatePartial

    if


    DISTINCT merge/terminatePartial


    if


    NO

  212. DISTINCT (mode = MERGEPARTIAL)


  214. （

    4


    ）ReduceSinkOperator：Partitioining Key: grouping key。Sorting

  215. Key: grouping key

    if


    no DISTINCT grouping + distinct key


    if


    DISTINCT


  217. （

    5


    ）GroupByOperator：GroupByDesc.Mode.FINAL，Reducer: merge/terminate (mode = FINAL)


  219. }






  225. ReduceSinkOperator的processOp(Object row,

    int

  226. tag)会根据相应的条件设置Key的hash值，如第四种情况的第一个ReduceSinkOperator：Partitioning Key:

  227. random()

    if


    no DISTINCT grouping + distinct key


    if

  228. DISTINCT，如果没有DISTINCT字段，那么在OutputCollector.collect前会设置当前Key的hash值为一个随机

  229. 数，random =

    new


    Random(


    12345


    );。如果有DISTINCT字段，那么key的hash值跟grouping +

  230. distinct key有关。








  238. GroupByOperator：


  240. initializeOp(Configuration hconf)


  242. processOp(Object row,

    int


    tag)


  244. closeOp(

    boolean


    abort)


  246. forward(ArrayList<Object> keys, AggregationBuffer[] aggs)






  252. groupby10.q   groupby11.q


  254. set hive.map.aggr=

    false


    ;


  256. set hive.groupby.skewindata=

    false


    ;




  260. EXPLAIN


  262. FROM INPUT


  264. INSERT OVERWRITE TABLE dest1 SELECT INPUT.key,

  265. count(substr(INPUT.value,

    5


    )), count(distinct substr(INPUT.value,


    5


    ))

  266. GROUP BY INPUT.key;




  270. STAGE DEPENDENCIES:


  272. Stage-

    1


    is a root stage


  274. Stage-

    0


    depends on stages: Stage-


    1




  278. STAGE PLANS:


  280. Stage: Stage-

    1


  282. Map Reduce


  284. Alias -> Map Operator Tree:


  286. input


  288. TableScan


  290. alias: input


  292. Select Operator

    // insertSelectAllPlanForGroupBy


  294. expressions:


  296. expr: key


  298. type:

    int


  300. expr: value


  302. type: string


  304. outputColumnNames: key, value


  306. Reduce Output Operator


  308. key expressions:


  310. expr: key


  312. type:

    int


  314. expr: substr(value,

    5


    )


  316. type: string


  318. sort order: ++


  320. Map-reduce partition columns:


  322. expr: key


  324. type:

    int


  326. tag: –

    1


  328. Reduce Operator Tree:


  330. Group By Operator


  332. aggregations:


  334. expr: count(KEY._col1:

    0


    ._col0)


  336. expr: count(DISTINCT KEY._col1:

    0


    ._col0)


  338. bucketGroup:

    false


  340. keys:


  342. expr: KEY._col0


  344. type:

    int


  346. mode: complete


  348. outputColumnNames: _col0, _col1, _col2


  350. Select Operator


  352. expressions:


  354. expr: _col0


  356. type:

    int


  358. expr: _col1


  360. type: bigint


  362. expr: _col2


  364. type: bigint


  366. outputColumnNames: _col0, _col1, _col2


  368. Select Operator


  370. expressions:


  372. expr: _col0


  374. type:

    int


  376. expr: UDFToInteger(_col1)


  378. type:

    int


  380. expr: UDFToInteger(_col2)


  382. type:

    int


  384. outputColumnNames: _col0, _col1, _col2


  386. File Output Operator


  388. compressed:

    false


  390. GlobalTableId:

    1


  392. table:


  394. input format: org.apache.hadoop.mapred.TextInputFormat


  396. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  398. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  400. name: dest1




  404. Stage: Stage-

    0


  406. Move Operator


  408. tables:


  410. replace:

    true


  412. table:


  414. input format: org.apache.hadoop.mapred.TextInputFormat


  416. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  418. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  420. name: dest1










  430. set hive.map.aggr=

    true


    ;


  432. set hive.groupby.skewindata=

    false


    ;




  436. STAGE DEPENDENCIES:


  438. Stage-

    1


    is a root stage


  440. Stage-

    0


    depends on stages: Stage-


    1




  444. STAGE PLANS:


  446. Stage: Stage-

    1


  448. Map Reduce


  450. Alias -> Map Operator Tree:


  452. input


  454. TableScan


  456. alias: input


  458. Select Operator


  460. expressions:


  462. expr: key


  464. type:

    int


  466. expr: value


  468. type: string


  470. outputColumnNames: key, value


  472. Group By Operator


  474. aggregations:


  476. expr: count(substr(value,

    5


    ))


  478. expr: count(DISTINCT substr(value,

    5


    ))


  480. bucketGroup:

    false


  482. keys:


  484. expr: key


  486. type:

    int


  488. expr: substr(value,

    5


    )


  490. type: string


  492. mode: hash


  494. outputColumnNames: _col0, _col1, _col2, _col3


  496. Reduce Output Operator


  498. key expressions:


  500. expr: _col0


  502. type:

    int


  504. expr: _col1


  506. type: string


  508. sort order: ++


  510. Map-reduce partition columns:


  512. expr: _col0


  514. type:

    int


  516. tag: –

    1


  518. value expressions:


  520. expr: _col2


  522. type: bigint


  524. expr: _col3


  526. type: bigint


  528. Reduce Operator Tree:


  530. Group By Operator


  532. aggregations:


  534. expr: count(VALUE._col0)


  536. expr: count(DISTINCT KEY._col1:

    0


    ._col0)


  538. bucketGroup:

    false


  540. keys:


  542. expr: KEY._col0


  544. type:

    int


  546. mode: mergepartial


  548. outputColumnNames: _col0, _col1, _col2


  550. Select Operator


  552. expressions:


  554. expr: _col0


  556. type:

    int


  558. expr: _col1


  560. type: bigint


  562. expr: _col2


  564. type: bigint


  566. outputColumnNames: _col0, _col1, _col2


  568. Select Operator


  570. expressions:


  572. expr: _col0


  574. type:

    int


  576. expr: UDFToInteger(_col1)


  578. type:

    int


  580. expr: UDFToInteger(_col2)


  582. type:

    int


  584. outputColumnNames: _col0, _col1, _col2


  586. File Output Operator


  588. compressed:

    false


  590. GlobalTableId:

    1


  592. table:


  594. input format: org.apache.hadoop.mapred.TextInputFormat


  596. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  598. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  600. name: dest1




  604. Stage: Stage-

    0


  606. Move Operator


  608. tables:


  610. replace:

    true


  612. table:


  614. input format: org.apache.hadoop.mapred.TextInputFormat


  616. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  618. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  620. name: dest1


















  638. set hive.map.aggr=

    false


    ;


  640. set hive.groupby.skewindata=

    true


    ;




  644. STAGE DEPENDENCIES:


  646. Stage-

    1


    is a root stage


  648. Stage-

    2


    depends on stages: Stage-


    1


  650. Stage-

    0


    depends on stages: Stage-


    2




  654. STAGE PLANS:


  656. Stage: Stage-

    1


  658. Map Reduce


  660. Alias -> Map Operator Tree:


  662. input


  664. TableScan


  666. alias: input


  668. Select Operator


  670. expressions:


  672. expr: key


  674. type:

    int


  676. expr: value


  678. type: string


  680. outputColumnNames: key, value


  682. Reduce Output Operator


  684. key expressions:


  686. expr: key


  688. type:

    int


  690. expr: substr(value,

    5


    )


  692. type: string


  694. sort order: ++


  696. Map-reduce partition columns:


  698. expr: key


  700. type:

    int


  702. tag: –

    1


  704. Reduce Operator Tree:


  706. Group By Operator


  708. aggregations:


  710. expr: count(KEY._col1:

    0


    ._col0)


  712. expr: count(DISTINCT KEY._col1:

    0


    ._col0)


  714. bucketGroup:

    false


  716. keys:


  718. expr: KEY._col0


  720. type:

    int


  722. mode: partial1


  724. outputColumnNames: _col0, _col1, _col2


  726. File Output Operator


  728. compressed:

    false


  730. GlobalTableId:

    0


  732. table:


  734. input format: org.apache.hadoop.mapred.SequenceFileInputFormat


  736. output format: org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat




  740. Stage: Stage-

    2


  742. Map Reduce


  744. Alias -> Map Operator Tree:


  746. hdfs:

    //localhost:54310/tmp/hive-tianzhao/hive_2011-07-15_21-48-26_387_7978992474997402829/-mr-10002


  748. Reduce Output Operator


  750. key expressions:


  752. expr: _col0


  754. type:

    int


  756. sort order: +


  758. Map-reduce partition columns:


  760. expr: _col0


  762. type:

    int


  764. tag: –

    1


  766. value expressions:


  768. expr: _col1


  770. type: bigint


  772. expr: _col2


  774. type: bigint


  776. Reduce Operator Tree:


  778. Group By Operator


  780. aggregations:


  782. expr: count(VALUE._col0)


  784. expr: count(VALUE._col1)


  786. bucketGroup:

    false


  788. keys:


  790. expr: KEY._col0


  792. type:

    int


  794. mode:

    final


  796. outputColumnNames: _col0, _col1, _col2


  798. Select Operator


  800. expressions:


  802. expr: _col0


  804. type:

    int


  806. expr: _col1


  808. type: bigint


  810. expr: _col2


  812. type: bigint


  814. outputColumnNames: _col0, _col1, _col2


  816. Select Operator


  818. expressions:


  820. expr: _col0


  822. type:

    int


  824. expr: UDFToInteger(_col1)


  826. type:

    int


  828. expr: UDFToInteger(_col2)


  830. type:

    int


  832. outputColumnNames: _col0, _col1, _col2


  834. File Output Operator


  836. compressed:

    false


  838. GlobalTableId:

    1


  840. table:


  842. input format: org.apache.hadoop.mapred.TextInputFormat


  844. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  846. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  848. name: dest1




  852. Stage: Stage-

    0


  854. Move Operator


  856. tables:


  858. replace:

    true


  860. table:


  862. input format: org.apache.hadoop.mapred.TextInputFormat


  864. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  866. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  868. name: dest1










  878. set hive.map.aggr=

    true


    ;


  880. set hive.groupby.skewindata=

    true


    ;




  884. STAGE DEPENDENCIES:


  886. Stage-

    1


    is a root stage


  888. Stage-

    2


    depends on stages: Stage-


    1


  890. Stage-

    0


    depends on stages: Stage-


    2




  894. STAGE PLANS:


  896. Stage: Stage-

    1


  898. Map Reduce


  900. Alias -> Map Operator Tree:


  902. input


  904. TableScan


  906. alias: input


  908. Select Operator


  910. expressions:


  912. expr: key


  914. type:

    int


  916. expr: value


  918. type: string


  920. outputColumnNames: key, value


  922. Group By Operator


  924. aggregations:


  926. expr: count(substr(value,

    5


    ))


  928. expr: count(DISTINCT substr(value,

    5


    ))


  930. bucketGroup:

    false


  932. keys:


  934. expr: key


  936. type:

    int


  938. expr: substr(value,

    5


    )


  940. type: string


  942. mode: hash


  944. outputColumnNames: _col0, _col1, _col2, _col3


  946. Reduce Output Operator


  948. key expressions:


  950. expr: _col0


  952. type:

    int


  954. expr: _col1


  956. type: string


  958. sort order: ++


  960. Map-reduce partition columns:


  962. expr: _col0


  964. type:

    int


  966. tag: –

    1


  968. value expressions:


  970. expr: _col2


  972. type: bigint


  974. expr: _col3


  976. type: bigint


  978. Reduce Operator Tree:


  980. Group By Operator


  982. aggregations:


  984. expr: count(VALUE._col0)


  986. expr: count(DISTINCT KEY._col1:

    0


    ._col0)


  988. bucketGroup:

    false


  990. keys:


  992. expr: KEY._col0


  994. type:

    int


  996. mode: partials


  998. outputColumnNames: _col0, _col1, _col2


  1000. File Output Operator


  1002. compressed:

    false


  1004. GlobalTableId:

    0


  1006. table:


  1008. input format: org.apache.hadoop.mapred.SequenceFileInputFormat


  1010. output format: org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat




  1014. Stage: Stage-

    2


  1016. Map Reduce


  1018. Alias -> Map Operator Tree:


  1020. hdfs:

    //localhost:54310/tmp/hive-tianzhao/hive_2011-07-15_21-49-25_899_4946067838822964610/-mr-10002


  1022. Reduce Output Operator


  1024. key expressions:


  1026. expr: _col0


  1028. type:

    int


  1030. sort order: +


  1032. Map-reduce partition columns:


  1034. expr: _col0


  1036. type:

    int


  1038. tag: –

    1


  1040. value expressions:


  1042. expr: _col1


  1044. type: bigint


  1046. expr: _col2


  1048. type: bigint


  1050. Reduce Operator Tree:


  1052. Group By Operator


  1054. aggregations:


  1056. expr: count(VALUE._col0)


  1058. expr: count(VALUE._col1)


  1060. bucketGroup:

    false


  1062. keys:


  1064. expr: KEY._col0


  1066. type:

    int


  1068. mode:

    final


  1070. outputColumnNames: _col0, _col1, _col2


  1072. Select Operator


  1074. expressions:


  1076. expr: _col0


  1078. type:

    int


  1080. expr: _col1


  1082. type: bigint


  1084. expr: _col2


  1086. type: bigint


  1088. outputColumnNames: _col0, _col1, _col2


  1090. Select Operator


  1092. expressions:


  1094. expr: _col0


  1096. type:

    int


  1098. expr: UDFToInteger(_col1)


  1100. type:

    int


  1102. expr: UDFToInteger(_col2)


  1104. type:

    int


  1106. outputColumnNames: _col0, _col1, _col2


  1108. File Output Operator


  1110. compressed:

    false


  1112. GlobalTableId:

    1


  1114. table:


  1116. input format: org.apache.hadoop.mapred.TextInputFormat


  1118. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1120. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1122. name: dest1




  1126. Stage: Stage-

    0


  1128. Move Operator


  1130. tables:


  1132. replace:

    true


  1134. table:


  1136. input format: org.apache.hadoop.mapred.TextInputFormat


  1138. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1140. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1142. name: dest1


















  1160. set hive.map.aggr=

    false


    ;


  1162. set hive.groupby.skewindata=

    false


    ;




  1166. EXPLAIN extended


  1168. FROM INPUT


  1170. INSERT OVERWRITE TABLE dest1 SELECT INPUT.key,

  1171. count(substr(INPUT.value,

    5


    )), count(distinct substr(INPUT.value,


    5


    ))

  1172. GROUP BY INPUT.key;




  1176. STAGE DEPENDENCIES:


  1178. Stage-

    1


    is a root stage


  1180. Stage-

    0


    depends on stages: Stage-


    1




  1184. STAGE PLANS:


  1186. Stage: Stage-

    1


  1188. Map Reduce


  1190. Alias -> Map Operator Tree:


  1192. input


  1194. TableScan


  1196. alias: input


  1198. Select Operator


  1200. expressions:


  1202. expr: key


  1204. type:

    int


  1206. expr: value


  1208. type: string


  1210. outputColumnNames: key, value


  1212. Reduce Output Operator


  1214. key expressions:


  1216. expr: key


  1218. type:

    int


  1220. expr: substr(value,

    5


    )


  1222. type: string


  1224. sort order: ++


  1226. Map-reduce partition columns:


  1228. expr: key


  1230. type:

    int


  1232. tag: –

    1


  1234. Needs Tagging:

    false


  1236. Path -> Alias:


  1238. hdfs:

    //localhost:54310/user/hive/warehouse/input [input]


  1240. Path -> Partition:


  1242. hdfs:

    //localhost:54310/user/hive/warehouse/input


  1244. Partition


  1246. base file name: input


  1248. input format: org.apache.hadoop.mapred.TextInputFormat


  1250. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1252. properties:


  1254. bucket_count –

    1


  1256. columns key,value


  1258. columns.types

    int


    :string


  1260. file.inputformat org.apache.hadoop.mapred.TextInputFormat


  1262. file.outputformat org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1264. location hdfs:

    //localhost:54310/user/hive/warehouse/input


  1266. name input


  1268. serialization.ddl struct input { i32 key, string value}


  1270. serialization.format

    1


  1272. serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1274. transient_lastDdlTime

    1310523947


  1276. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe




  1280. input format: org.apache.hadoop.mapred.TextInputFormat


  1282. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1284. properties:


  1286. bucket_count –

    1


  1288. columns key,value


  1290. columns.types

    int


    :string


  1292. file.inputformat org.apache.hadoop.mapred.TextInputFormat


  1294. file.outputformat org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1296. location hdfs:

    //localhost:54310/user/hive/warehouse/input


  1298. name input


  1300. serialization.ddl struct input { i32 key, string value}


  1302. serialization.format

    1


  1304. serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1306. transient_lastDdlTime

    1310523947


  1308. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1310. name: input


  1312. name: input


  1314. Reduce Operator Tree:


  1316. Group By Operator


  1318. aggregations:


  1320. expr: count(KEY._col1:

    0


    ._col0)


  1322. expr: count(DISTINCT KEY._col1:

    0


    ._col0)


  1324. bucketGroup:

    false


  1326. keys:


  1328. expr: KEY._col0


  1330. type:

    int


  1332. mode: complete


  1334. outputColumnNames: _col0, _col1, _col2


  1336. Select Operator


  1338. expressions:


  1340. expr: _col0


  1342. type:

    int


  1344. expr: _col1


  1346. type: bigint


  1348. expr: _col2


  1350. type: bigint


  1352. outputColumnNames: _col0, _col1, _col2


  1354. Select Operator


  1356. expressions:


  1358. expr: _col0


  1360. type:

    int


  1362. expr: UDFToInteger(_col1)


  1364. type:

    int


  1366. expr: UDFToInteger(_col2)


  1368. type:

    int


  1370. outputColumnNames: _col0, _col1, _col2


  1372. File Output Operator


  1374. compressed:

    false


  1376. GlobalTableId:

    1


  1378. directory: hdfs:

    //localhost:54310/tmp/hive-tianzhao/hive_2011-07-15_21-50-38_510_6852880850328147221/-ext-10000


  1380. NumFilesPerFileSink:

    1


  1382. table:


  1384. input format: org.apache.hadoop.mapred.TextInputFormat


  1386. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1388. properties:


  1390. bucket_count –

    1


  1392. columns key,val1,val2


  1394. columns.types

    int


    :


    int


    :


    int


  1396. file.inputformat org.apache.hadoop.mapred.TextInputFormat


  1398. file.outputformat org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1400. location hdfs:

    //localhost:54310/user/hive/warehouse/dest1


  1402. name dest1


  1404. serialization.ddl struct dest1 { i32 key, i32 val1, i32 val2}


  1406. serialization.format

    1


  1408. serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1410. transient_lastDdlTime

    1310523946


  1412. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1414. name: dest1


  1416. TotalFiles:

    1


  1418. MultiFileSpray:

    false




  1422. Stage: Stage-

    0


  1424. Move Operator


  1426. tables:


  1428. replace:

    true


  1430. source: hdfs:

    //localhost:54310/tmp/hive-tianzhao/hive_2011-07-15_21-50-38_510_6852880850328147221/-ext-10000


  1432. table:


  1434. input format: org.apache.hadoop.mapred.TextInputFormat


  1436. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1438. properties:


  1440. bucket_count –

    1


  1442. columns key,val1,val2


  1444. columns.types

    int


    :


    int


    :


    int


  1446. file.inputformat org.apache.hadoop.mapred.TextInputFormat


  1448. file.outputformat org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1450. location hdfs:

    //localhost:54310/user/hive/warehouse/dest1


  1452. name dest1


  1454. serialization.ddl struct dest1 { i32 key, i32 val1, i32 val2}


  1456. serialization.format

    1


  1458. serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1460. transient_lastDdlTime

    1310523946


  1462. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1464. name: dest1


  1466. tmp directory: hdfs:

    //localhost:54310/tmp/hive-tianzhao/hive_2011-07-15_21-50-38_510_6852880850328147221/-ext-10001








  1474. ABSTRACT SYNTAX TREE:


  1476. (TOK_QUERY


  1478. (TOK_FROM (TOK_TABREF INPUT))


  1480. (TOK_INSERT


  1482. (TOK_DESTINATION (TOK_TAB dest1))


  1484. (TOK_SELECT


  1486. (TOK_SELEXPR (. (TOK_TABLE_OR_COL INPUT) key))


  1488. (TOK_SELEXPR (TOK_FUNCTION count (TOK_FUNCTION substr (. (TOK_TABLE_OR_COL INPUT) value)

    5


    )))


  1490. (TOK_SELEXPR (TOK_FUNCTIONDI count (TOK_FUNCTION substr (. (TOK_TABLE_OR_COL INPUT) value)

    5


    )))


  1492. )


  1494. (TOK_GROUPBY (. (TOK_TABLE_OR_COL INPUT) key))


  1496. )


  1498. )








  1506. SemanticAnalyzer.genBodyPlan(QB qb, Operator input){



  1508. if


    (qbp.getAggregationExprsForClause(dest).size() !=


    0


  1510. || getGroupByForClause(qbp, dest).size() >

    0


    ) {


    //如果有聚合函数或者有groupby，则执行下面的操作



  1512. //multiple distincts is not supported with skew in data



  1514. if


    (conf.getVar(HiveConf.ConfVars.HIVEGROUPBYSKEW)


  1516. .equalsIgnoreCase(

    “true”


    ) &&


  1518. qbp.getDistinctFuncExprsForClause(dest).size() >

    1


    ) {



  1520. throw




    new


    SemanticException(ErrorMsg.UNSUPPORTED_MULTIPLE_DISTINCTS.


  1522. getMsg());


  1524. }



  1526. // insert a select operator here used by the ColumnPruner to reduce



  1528. // the data to shuffle


  1530. curr = insertSelectAllPlanForGroupBy(dest, curr);

    //生成一个SelectOperator，所有的字段都会选取，selectStar=true。



  1532. if


    (conf.getVar(HiveConf.ConfVars.HIVEMAPSIDEAGGREGATE)


  1534. .equalsIgnoreCase(

    “true”


    )) {



  1536. if


    (conf.getVar(HiveConf.ConfVars.HIVEGROUPBYSKEW)


  1538. .equalsIgnoreCase(

    “false”


    )) {


  1540. curr = genGroupByPlanMapAggr1MR(dest, qb, curr);


  1542. }

    else


    {


  1544. curr = genGroupByPlanMapAggr2MR(dest, qb, curr);


  1546. }


  1548. }

    else




    if


    (conf.getVar(HiveConf.ConfVars.HIVEGROUPBYSKEW)


  1550. .equalsIgnoreCase(

    “true”


    )) {


  1552. curr = genGroupByPlan2MR(dest, qb, curr);


  1554. }

    else


    {


  1556. curr = genGroupByPlan1MR(dest, qb, curr);


  1558. }


  1560. }


  1562. }




  1566. distince：


  1568. count.q.out


  1570. groupby11.q.out


  1572. groupby10.q.out


  1574. nullgroup4_multi_distinct.q.out


  1576. join18.q.out


  1578. groupby_bigdata.q.out


  1580. join18_multi_distinct.q.out


  1582. nullgroup4.q.out


  1584. auto_join18_multi_distinct.q.out


  1586. auto_join18.q.out




  1590. （

    1


    ）map端部分聚合，数据无倾斜，一个MR生成。


  1592. genGroupByPlanMapAggr1MR，生成三个Operator：


  1594. （

    1.1


    ）GroupByOperator：map-side partial aggregation，由genGroupByPlanMapGroupByOperator方法生成：


  1596. 处理groupby子句，getGroupByForClause，groupby的column加入groupByKeys和outputColumnNames


  1598. 处理select中的Distinct，getDistinctFuncExprsForClause，Distinct的column，加入groupByKeys和outputColumnNames


  1600. 处理聚合函数，getAggregationExprsForClause，生成AggregationDesc加入aggregations，生成column加入outputColumnNames



  1602. public


    GroupByDesc(



  1604. final


    Mode mode,



  1606. final


    java.util.ArrayList<java.lang.String> outputColumnNames,



  1608. final


    java.util.ArrayList<ExprNodeDesc> keys,



  1610. final


    java.util.ArrayList<org.apache.hadoop.hive.ql.plan.AggregationDesc> aggregators,



  1612. final




    boolean


    groupKeyNotReductionKey,


    float


    groupByMemoryUsage,


    float


    memoryThreshold) {



  1614. this


    (mode, outputColumnNames, keys, aggregators, groupKeyNotReductionKey,



  1616. false


    , groupByMemoryUsage, memoryThreshold);


  1618. }


  1620. mode：GroupByDesc.Mode.HASH


  1622. outputColumnNames：groupby+Distinct+Aggregation


  1624. keys：groupby+Distinct


  1626. aggregators：Aggregation


  1628. groupKeyNotReductionKey：

    false


  1630. groupByMemoryUsage：默认为

    0.5


  1632. memoryThreshold：默认为

    0.9




  1636. （

    1.2


    ）ReduceSinkOperator


  1638. 处理groupby子句，getGroupByForClause，groupby的column加入reduceKeys和outputKeyColumnNames


  1640. 处理select中的Distinct，getDistinctFuncExprsForClause，Distinct的column，加入reduceKeys和outputKeyColumnNames


  1642. 处理聚合函数，getAggregationExprsForClause，需要做聚合的column加入reduceValues和outputValueColumnNames



  1644. public


    ReduceSinkDesc(java.util.ArrayList<ExprNodeDesc> keyCols,



  1646. int


    numDistributionKeys,


  1648. java.util.ArrayList<ExprNodeDesc> valueCols,


  1650. java.util.ArrayList<java.lang.String> outputKeyColumnNames,


  1652. List<List<Integer>> distinctColumnIndices,


  1654. java.util.ArrayList<java.lang.String> outputValueColumnNames,

    int


    tag,


  1656. java.util.ArrayList<ExprNodeDesc> partitionCols,

    int


    numReducers,



  1658. final


    TableDesc keySerializeInfo,


    final


    TableDesc valueSerializeInfo) {



  1660. this


    .keyCols = keyCols;


    // 为reduceKeys，groupby+distinct



  1662. this


    .numDistributionKeys = numDistributionKeys;


    // grpByExprs.size()



  1664. this


    .valueCols = valueCols;


    //reduceValues，聚合函数



  1666. this


    .outputKeyColumnNames = outputKeyColumnNames;


    //outputKeyColumnNames



  1668. this


    .outputValueColumnNames = outputValueColumnNames;


    //outputValueColumnNames



  1670. this


    .tag = tag;


    // -1



  1672. this


    .numReducers = numReducers;


    // 一般都是-1



  1674. this


    .partitionCols = partitionCols;


    // groupby



  1676. this


    .keySerializeInfo = keySerializeInfo;



  1678. this


    .valueSerializeInfo = valueSerializeInfo;



  1680. this


    .distinctColumnIndices = distinctColumnIndices;


  1682. }




  1686. （

    1.3


    ）GroupByOperator


  1688. 处理groupby子句，getGroupByForClause，groupby的column加入reduceKeys和outputKeyColumnNames


  1690. 处理聚合函数，getAggregationExprsForClause，需要做聚合的column加入reduceValues和outputValueColumnNames



  1692. public


    GroupByDesc(



  1694. final


    Mode mode,



  1696. final


    java.util.ArrayList<java.lang.String> outputColumnNames,



  1698. final


    java.util.ArrayList<ExprNodeDesc> keys,



  1700. final


    java.util.ArrayList<org.apache.hadoop.hive.ql.plan.AggregationDesc> aggregators,



  1702. final




    boolean


    groupKeyNotReductionKey,


    float


    groupByMemoryUsage,


    float


    memoryThreshold) {



  1704. this


    (mode, outputColumnNames, keys, aggregators, groupKeyNotReductionKey,



  1706. false


    , groupByMemoryUsage, memoryThreshold);


  1708. }


  1710. mode：GroupByDesc.Mode.MERGEPARTIAL


  1712. outputColumnNames：groupby+Aggregation


  1714. keys：groupby


  1716. aggregators：Aggregation


  1718. groupKeyNotReductionKey：

    false


  1720. groupByMemoryUsage：默认为

    0.5


  1722. memoryThreshold：默认为

    0.9




  1726. （

    2


    ）map端部分聚合，数据倾斜，两个MR生成。


  1728. genGroupByPlanMapAggr2MR：


  1730. （

    2.1


    ）GroupByOperator：map-side partial aggregation，由genGroupByPlanMapGroupByOperator方法生成：


  1732. 处理groupby子句，getGroupByForClause，groupby的column加入groupByKeys和outputColumnNames


  1734. 处理select中的Distinct，getDistinctFuncExprsForClause，Distinct的column，加入groupByKeys和outputColumnNames


  1736. 处理聚合函数，getAggregationExprsForClause，生成AggregationDesc加入aggregations，生成column加入outputColumnNames



  1738. public


    GroupByDesc(



  1740. final


    Mode mode,



  1742. final


    java.util.ArrayList<java.lang.String> outputColumnNames,



  1744. final


    java.util.ArrayList<ExprNodeDesc> keys,



  1746. final


    java.util.ArrayList<org.apache.hadoop.hive.ql.plan.AggregationDesc> aggregators,



  1748. final




    boolean


    groupKeyNotReductionKey,


    float


    groupByMemoryUsage,


    float


    memoryThreshold) {



  1750. this


    (mode, outputColumnNames, keys, aggregators, groupKeyNotReductionKey,



  1752. false


    , groupByMemoryUsage, memoryThreshold);


  1754. }


  1756. mode：GroupByDesc.Mode.HASH


  1758. outputColumnNames：groupby+Distinct+Aggregation


  1760. keys：groupby+Distinct


  1762. aggregators：Aggregation


  1764. groupKeyNotReductionKey：

    false


  1766. groupByMemoryUsage：默认为

    0.5


  1768. memoryThreshold：默认为

    0.9




  1772. （

    2.2


    ）ReduceSinkOperator


  1774. 处理groupby子句，getGroupByForClause，groupby的column加入reduceKeys和outputKeyColumnNames


  1776. 处理select中的Distinct，getDistinctFuncExprsForClause，Distinct的column，加入reduceKeys和outputKeyColumnNames


  1778. 处理聚合函数，getAggregationExprsForClause，需要做聚合的column加入reduceValues和outputValueColumnNames



  1780. public


    ReduceSinkDesc(java.util.ArrayList<ExprNodeDesc> keyCols,



  1782. int


    numDistributionKeys,


  1784. java.util.ArrayList<ExprNodeDesc> valueCols,


  1786. java.util.ArrayList<java.lang.String> outputKeyColumnNames,


  1788. List<List<Integer>> distinctColumnIndices,


  1790. java.util.ArrayList<java.lang.String> outputValueColumnNames,

    int


    tag,


  1792. java.util.ArrayList<ExprNodeDesc> partitionCols,

    int


    numReducers,



  1794. final


    TableDesc keySerializeInfo,


    final


    TableDesc valueSerializeInfo) {



  1796. this


    .keyCols = keyCols;


    // 为reduceKeys，groupby+distinct



  1798. this


    .numDistributionKeys = numDistributionKeys;


    // grpByExprs.size()



  1800. this


    .valueCols = valueCols;


    //reduceValues，聚合函数



  1802. this


    .outputKeyColumnNames = outputKeyColumnNames;


    //outputKeyColumnNames



  1804. this


    .outputValueColumnNames = outputValueColumnNames;


    //outputValueColumnNames



  1806. this


    .tag = tag;


    // -1



  1808. this


    .numReducers = numReducers;


    // 一般都是-1



  1810. this


    .partitionCols = partitionCols;


    // groupby



  1812. this


    .keySerializeInfo = keySerializeInfo;



  1814. this


    .valueSerializeInfo = valueSerializeInfo;



  1816. this


    .distinctColumnIndices = distinctColumnIndices;


  1818. }




  1822. （

    2.3


    ）GroupByOperator


  1824. 处理groupby子句，getGroupByForClause，groupby的column加入groupByKeys和outputColumnNames


  1826. 处理聚合函数，getAggregationExprsForClause，生成AggregationDesc加入aggregations，生成column加入outputColumnNames



  1828. public


    GroupByDesc(



  1830. final


    Mode mode,



  1832. final


    java.util.ArrayList<java.lang.String> outputColumnNames,



  1834. final


    java.util.ArrayList<ExprNodeDesc> keys,



  1836. final


    java.util.ArrayList<org.apache.hadoop.hive.ql.plan.AggregationDesc> aggregators,



  1838. final




    boolean


    groupKeyNotReductionKey,


    float


    groupByMemoryUsage,


    float


    memoryThreshold) {



  1840. this


    (mode, outputColumnNames, keys, aggregators, groupKeyNotReductionKey,



  1842. false


    , groupByMemoryUsage, memoryThreshold);


  1844. }


  1846. mode：GroupByDesc.Mode.PARTIALS


  1848. outputColumnNames：groupby+Aggregation


  1850. keys：groupby


  1852. aggregators：Aggregation


  1854. groupKeyNotReductionKey：

    false


  1856. groupByMemoryUsage：默认为

    0.5


  1858. memoryThreshold：默认为

    0.9




  1862. （

    2.4


    ）ReduceSinkOperator


  1864. 处理groupby子句，getGroupByForClause，groupby的column加入reduceKeys和outputColumnNames


  1866. 处理聚合函数，getAggregationExprsForClause，需要做聚合的column加入reduceValues和outputColumnNames



  1868. public


    ReduceSinkDesc(java.util.ArrayList<ExprNodeDesc> keyCols,



  1870. int


    numDistributionKeys,


  1872. java.util.ArrayList<ExprNodeDesc> valueCols,


  1874. java.util.ArrayList<java.lang.String> outputKeyColumnNames,


  1876. List<List<Integer>> distinctColumnIndices,


  1878. java.util.ArrayList<java.lang.String> outputValueColumnNames,

    int


    tag,


  1880. java.util.ArrayList<ExprNodeDesc> partitionCols,

    int


    numReducers,



  1882. final


    TableDesc keySerializeInfo,


    final


    TableDesc valueSerializeInfo) {



  1884. this


    .keyCols = keyCols;


    // 为reduceKeys，groupby



  1886. this


    .numDistributionKeys = numDistributionKeys;


    // grpByExprs.size()



  1888. this


    .valueCols = valueCols;


    //reduceValues，聚合函数



  1890. this


    .outputKeyColumnNames = outputKeyColumnNames;


    //outputKeyColumnNames



  1892. this


    .outputValueColumnNames = outputValueColumnNames;


    //outputValueColumnNames



  1894. this


    .tag = tag;


    // -1



  1896. this


    .numReducers = numReducers;


    // 一般都是-1



  1898. this


    .partitionCols = partitionCols;


    // groupby



  1900. this


    .keySerializeInfo = keySerializeInfo;



  1902. this


    .valueSerializeInfo = valueSerializeInfo;



  1904. this


    .distinctColumnIndices = distinctColumnIndices;


  1906. }




  1910. （

    2.5


    ）GroupByOperator


  1912. 处理groupby子句，getGroupByForClause，groupby的column加入groupByKeys和outputColumnNames


  1914. 处理聚合函数，getAggregationExprsForClause，生成AggregationDesc加入aggregations，需要做聚合的column加入outputColumnNames



  1916. public


    GroupByDesc(



  1918. final


    Mode mode,



  1920. final


    java.util.ArrayList<java.lang.String> outputColumnNames,



  1922. final


    java.util.ArrayList<ExprNodeDesc> keys,



  1924. final


    java.util.ArrayList<org.apache.hadoop.hive.ql.plan.AggregationDesc> aggregators,



  1926. final




    boolean


    groupKeyNotReductionKey,


    float


    groupByMemoryUsage,


    float


    memoryThreshold) {



  1928. this


    (mode, outputColumnNames, keys, aggregators, groupKeyNotReductionKey,



  1930. false


    , groupByMemoryUsage, memoryThreshold);


  1932. }


  1934. mode：GroupByDesc.Mode.FINAL


  1936. outputColumnNames：groupby+Aggregation


  1938. keys：groupby


  1940. aggregators：Aggregation


  1942. groupKeyNotReductionKey：

    false


  1944. groupByMemoryUsage：默认为

    0.5


  1946. memoryThreshold：默认为

    0.9






  1952. （

    3


    ）map端不部分聚合，数据倾斜，两个MR生成。


  1954. genGroupByPlan2MR：




  1958. （

    3.1


    ）ReduceSinkOperator


  1960. 处理groupby子句，getGroupByForClause，groupby的column加入reduceKeys和outputKeyColumnNames


  1962. 处理select中的Distinct，getDistinctFuncExprsForClause，Distinct的column，加入reduceKeys和outputKeyColumnNames


  1964. 处理聚合函数，getAggregationExprsForClause，需要做聚合的column加入reduceValues和outputValueColumnNames



  1966. public


    ReduceSinkDesc(java.util.ArrayList<ExprNodeDesc> keyCols,



  1968. int


    numDistributionKeys,


  1970. java.util.ArrayList<ExprNodeDesc> valueCols,


  1972. java.util.ArrayList<java.lang.String> outputKeyColumnNames,


  1974. List<List<Integer>> distinctColumnIndices,


  1976. java.util.ArrayList<java.lang.String> outputValueColumnNames,

    int


    tag,


  1978. java.util.ArrayList<ExprNodeDesc> partitionCols,

    int


    numReducers,



  1980. final


    TableDesc keySerializeInfo,


    final


    TableDesc valueSerializeInfo) {



  1982. this


    .keyCols = keyCols;


    // 为reduceKeys，groupby+distinct



  1984. this


    .numDistributionKeys = numDistributionKeys;


    // grpByExprs.size()



  1986. this


    .valueCols = valueCols;


    //reduceValues，聚合函数



  1988. this


    .outputKeyColumnNames = outputKeyColumnNames;


    //outputKeyColumnNames



  1990. this


    .outputValueColumnNames = outputValueColumnNames;


    //outputValueColumnNames



  1992. this


    .tag = tag;


    // -1



  1994. this


    .numReducers = numReducers;


    // 一般都是-1



  1996. this


    .partitionCols = partitionCols;


    // groupby



  1998. this


    .keySerializeInfo = keySerializeInfo;



  2000. this


    .valueSerializeInfo = valueSerializeInfo;



  2002. this


    .distinctColumnIndices = distinctColumnIndices;


  2004. }




  2008. （

    3.2


    ）GroupByOperator


  2010. 处理groupby子句，getGroupByForClause，groupby的column加入groupByKeys和outputColumnNames


  2012. 处理聚合函数，getAggregationExprsForClause，生成AggregationDesc加入aggregations，生成column加入outputColumnNames



  2014. public


    GroupByDesc(



  2016. final


    Mode mode,



  2018. final


    java.util.ArrayList<java.lang.String> outputColumnNames,



  2020. final


    java.util.ArrayList<ExprNodeDesc> keys,



  2022. final


    java.util.ArrayList<org.apache.hadoop.hive.ql.plan.AggregationDesc> aggregators,



  2024. final




    boolean


    groupKeyNotReductionKey,


    float


    groupByMemoryUsage,


    float


    memoryThreshold) {



  2026. this


    (mode, outputColumnNames, keys, aggregators, groupKeyNotReductionKey,



  2028. false


    , groupByMemoryUsage, memoryThreshold);


  2030. }


  2032. mode：GroupByDesc.Mode.PARTIAL1


  2034. outputColumnNames：groupby+Aggregation


  2036. keys：groupby


  2038. aggregators：Aggregation


  2040. groupKeyNotReductionKey：

    false


  2042. groupByMemoryUsage：默认为

    0.5


  2044. memoryThreshold：默认为

    0.9




  2048. （

    3.3


    ）ReduceSinkOperator


  2050. 处理groupby子句，getGroupByForClause，groupby的column加入reduceKeys和outputColumnNames


  2052. 处理聚合函数，getAggregationExprsForClause，需要做聚合的column加入reduceValues和outputColumnNames



  2054. public


    ReduceSinkDesc(java.util.ArrayList<ExprNodeDesc> keyCols,



  2056. int


    numDistributionKeys,


  2058. java.util.ArrayList<ExprNodeDesc> valueCols,


  2060. java.util.ArrayList<java.lang.String> outputKeyColumnNames,


  2062. List<List<Integer>> distinctColumnIndices,


  2064. java.util.ArrayList<java.lang.String> outputValueColumnNames,

    int


    tag,


  2066. java.util.ArrayList<ExprNodeDesc> partitionCols,

    int


    numReducers,



  2068. final


    TableDesc keySerializeInfo,


    final


    TableDesc valueSerializeInfo) {



  2070. this


    .keyCols = keyCols;


    // 为reduceKeys，groupby



  2072. this


    .numDistributionKeys = numDistributionKeys;


    // grpByExprs.size()



  2074. this


    .valueCols = valueCols;


    //reduceValues，聚合函数



  2076. this


    .outputKeyColumnNames = outputKeyColumnNames;


    //outputKeyColumnNames



  2078. this


    .outputValueColumnNames = outputValueColumnNames;


    //outputValueColumnNames



  2080. this


    .tag = tag;


    // -1



  2082. this


    .numReducers = numReducers;


    // 一般都是-1



  2084. this


    .partitionCols = partitionCols;


    // groupby



  2086. this


    .keySerializeInfo = keySerializeInfo;



  2088. this


    .valueSerializeInfo = valueSerializeInfo;



  2090. this


    .distinctColumnIndices = distinctColumnIndices;


  2092. }




  2096. （

    3.4


    ）GroupByOperator


  2098. 处理groupby子句，getGroupByForClause，groupby的column加入groupByKeys和outputColumnNames


  2100. 处理聚合函数，getAggregationExprsForClause，生成AggregationDesc加入aggregations，需要做聚合的column加入outputColumnNames



  2102. public


    GroupByDesc(



  2104. final


    Mode mode,



  2106. final


    java.util.ArrayList<java.lang.String> outputColumnNames,



  2108. final


    java.util.ArrayList<ExprNodeDesc> keys,



  2110. final


    java.util.ArrayList<org.apache.hadoop.hive.ql.plan.AggregationDesc> aggregators,



  2112. final




    boolean


    groupKeyNotReductionKey,


    float


    groupByMemoryUsage,


    float


    memoryThreshold) {



  2114. this


    (mode, outputColumnNames, keys, aggregators, groupKeyNotReductionKey,



  2116. false


    , groupByMemoryUsage, memoryThreshold);


  2118. }


  2120. mode：GroupByDesc.Mode.FINAL


  2122. outputColumnNames：groupby+Aggregation


  2124. keys：groupby


  2126. aggregators：Aggregation


  2128. groupKeyNotReductionKey：

    false


  2130. groupByMemoryUsage：默认为

    0.5


  2132. memoryThreshold：默认为

    0.9






  2138. （

    4


    ）map端不部分聚合，数据无倾斜，一个MR生成。


  2140. genGroupByPlan1MR：


  2142. （

    4.1


    ）ReduceSinkOperator


  2144. 处理groupby子句，getGroupByForClause，groupby的column加入reduceKeys和outputKeyColumnNames


  2146. 处理select中的Distinct，getDistinctFuncExprsForClause，Distinct的column，加入reduceKeys和outputKeyColumnNames


  2148. 处理聚合函数，getAggregationExprsForClause，需要做聚合的column加入reduceValues和outputValueColumnNames



  2150. public


    ReduceSinkDesc(java.util.ArrayList<ExprNodeDesc> keyCols,



  2152. int


    numDistributionKeys,


  2154. java.util.ArrayList<ExprNodeDesc> valueCols,


  2156. java.util.ArrayList<java.lang.String> outputKeyColumnNames,


  2158. List<List<Integer>> distinctColumnIndices,


  2160. java.util.ArrayList<java.lang.String> outputValueColumnNames,

    int


    tag,


  2162. java.util.ArrayList<ExprNodeDesc> partitionCols,

    int


    numReducers,



  2164. final


    TableDesc keySerializeInfo,


    final


    TableDesc valueSerializeInfo) {



  2166. this


    .keyCols = keyCols;


    // 为reduceKeys，groupby+distinct



  2168. this


    .numDistributionKeys = numDistributionKeys;


    // grpByExprs.size()



  2170. this


    .valueCols = valueCols;


    //reduceValues，聚合函数



  2172. this


    .outputKeyColumnNames = outputKeyColumnNames;


    //outputKeyColumnNames



  2174. this


    .outputValueColumnNames = outputValueColumnNames;


    //outputValueColumnNames



  2176. this


    .tag = tag;


    // -1



  2178. this


    .numReducers = numReducers;


    // 一般都是-1



  2180. this


    .partitionCols = partitionCols;


    // groupby



  2182. this


    .keySerializeInfo = keySerializeInfo;



  2184. this


    .valueSerializeInfo = valueSerializeInfo;



  2186. this


    .distinctColumnIndices = distinctColumnIndices;


  2188. }




  2192. （

    4.2


    ）GroupByOperator


  2194. 处理groupby子句，getGroupByForClause，groupby的column加入reduceKeys和outputKeyColumnNames


  2196. 处理聚合函数，getAggregationExprsForClause，需要做聚合的column加入reduceValues和outputValueColumnNames



  2198. public


    GroupByDesc(



  2200. final


    Mode mode,



  2202. final


    java.util.ArrayList<java.lang.String> outputColumnNames,



  2204. final


    java.util.ArrayList<ExprNodeDesc> keys,



  2206. final


    java.util.ArrayList<org.apache.hadoop.hive.ql.plan.AggregationDesc> aggregators,



  2208. final




    boolean


    groupKeyNotReductionKey,


    float


    groupByMemoryUsage,


    float


    memoryThreshold) {



  2210. this


    (mode, outputColumnNames, keys, aggregators, groupKeyNotReductionKey,



  2212. false


    , groupByMemoryUsage, memoryThreshold);


  2214. }


  2216. mode：GroupByDesc.Mode.COMPLETE


  2218. outputColumnNames：groupby+Aggregation


  2220. keys：groupby


  2222. aggregators：Aggregation


  2224. groupKeyNotReductionKey：

    false


  2226. groupByMemoryUsage：默认为

    0.5


  2228. memoryThreshold：默认为

    0.9








  2236. SemanticAnalyzer.genBodyPlan


  2238. optimizeMultiGroupBy  (multi-group by with the same distinct)


  2240. groupby10.q  groupby11.q



准备数据


[java]


view plain


copy



  1. SELECT uid, SUM(COUNT) FROM logs GROUP BY uid;


[java]


view plain


copy



  1. hive> SELECT * FROM logs;

  2. a   苹果

    5

  3. a   橙子

    3

  4. a      苹果

    2

  5. b   烧鸡

    1


  7. hive> SELECT uid, SUM(COUNT) FROM logs GROUP BY uid;

  8. a

    10

  9. b

    1



计算过程


hive-groupby-cal


默认设置了hive.map.aggr=true，所以会在mapper端先group by一次，最后再把结果merge起来，为了减少reducer处理的数据量。注意看explain的mode是不一样的。mapper是hash，reducer是mergepartial。如果把hive.map.aggr=false，那将groupby放到reducer才做，他的mode是complete.



Operator


hive-groupby-op



Explain


[java]


view plain


copy



  1. hive> explain SELECT uid, sum(count) FROM logs group by uid;

  2. OK

  3. ABSTRACT SYNTAX TREE:

  4. (TOK_QUERY (TOK_FROM (TOK_TABREF (TOK_TABNAME logs))) (TOK_INSERT (TOK_DESTINATION (TOK_DIR TOK_TMP_FILE)) (TOK_SELECT (TOK_SELEXPR (TOK_TABLE_OR_COL uid)) (TOK_SELEXPR (TOK_FUNCTION sum (TOK_TABLE_OR_COL count)))) (TOK_GROUPBY (TOK_TABLE_OR_COL uid))))


  6. STAGE DEPENDENCIES:

  7. Stage-

    1


    is a root stage

  8. Stage-

    0


    is a root stage


  10. STAGE PLANS:

  11. Stage: Stage-

    1

  12. Map Reduce

  13. Alias -> Map Operator Tree:

  14. logs

  15. TableScan

    // 扫描表

  16. alias: logs

  17. Select Operator

    //选择字段

  18. expressions:

  19. expr: uid

  20. type: string

  21. expr: count

  22. type:

    int

  23. outputColumnNames: uid, count

  24. Group By Operator

    //这里是因为默认设置了hive.map.aggr=true，会在mapper先做一次聚合，减少reduce需要处理的数据

  25. aggregations:

  26. expr: sum(count)

    //聚集函数

  27. bucketGroup:

    false

  28. keys:

    //键

  29. expr: uid

  30. type: string

  31. mode: hash

    //hash方式，processHashAggr()

  32. outputColumnNames: _col0, _col1

  33. Reduce Output Operator

    //输出key，value给reducer

  34. key expressions:

  35. expr: _col0

  36. type: string

  37. sort order: +

  38. Map-reduce partition columns:

  39. expr: _col0

  40. type: string

  41. tag: –

    1

  42. value expressions:

  43. expr: _col1

  44. type: bigint

  45. Reduce Operator Tree:

  46. Group By Operator


  48. aggregations:

  49. expr: sum(VALUE._col0)


  50. //聚合

  51. bucketGroup:

    false

  52. keys:

  53. expr: KEY._col0

  54. type: string

  55. mode: mergepartial

    //合并值

  56. outputColumnNames: _col0, _col1

  57. Select Operator

    //选择字段

  58. expressions:

  59. expr: _col0

  60. type: string

  61. expr: _col1

  62. type: bigint

  63. outputColumnNames: _col0, _col1

  64. File Output Operator

    //输出到文件

  65. compressed:

    false

  66. GlobalTableId:

    0

  67. table:

  68. input format: org.apache.hadoop.mapred.TextInputFormat

  69. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  71. Stage: Stage-

    0

  72. Fetch Operator

  73. limit: –

    1


  75. hive> select distinct value from src;

  76. hive> select max(key) from src;


  78. 因为没有grouping keys，所以只有一个reducer。







  84. 2.2


    如果有聚合函数或者groupby，做如下处理：


  86. 插入一个select operator，选取所有的字段，用于优化阶段ColumnPruner的优化



  88. 2.2


    .


    1


    hive.map.aggr为


    true


    ，默认是


    true


    ，开启的，在map端做部分聚合



  90. 2.2


    .


    1.1


    hive.groupby.skewindata为


    false


    ，默认是关闭的，groupby的数据没有倾斜。


  92. 生成的operator是: GroupByOperator+ReduceSinkOperator+GroupByOperator。


  94. GroupByOperator+ReduceSinkOperator用于在map端做操作，第一个GroupByOperator在map端先做部分聚合。第二个用于在reduce端做GroupBy操作



  96. 2.2


    .


    1.2


    hive.groupby.skewindata为


    true


  98. 生成的operator是: GroupbyOperator+ReduceSinkOperator+GroupbyOperator+ReduceSinkOperator +GroupByOperator


  100. GroupbyOperator+ReduceSinkOperator（第一个MapredTask的map阶段）


  102. GroupbyOperator(第一个MapredTask的reduce阶段)


  104. ReduceSinkOperator （第二个MapredTask的map阶段）


  106. GroupByOperator（第二个MapredTask的reduce阶段）



  108. 2.2


    .


    2


    hive.map.aggr为


    false



  110. 2.2


    .


    2.1


    hive.groupby.skewindata为


    true


  112. 生成的operator是: ReduceSinkOperator+GroupbyOperator+ReduceSinkOperator +GroupByOperator


  114. ReduceSinkOperator(第一个MapredTask的map阶段)


  116. GroupbyOperator(第一个MapredTask的reduce阶段)


  118. ReduceSinkOperator （第二个MapredTask的map阶段）


  120. GroupByOperator（第二个MapredTask的reduce阶段）



  122. 2.2


    .


    2.2


    hive.groupby.skewindata为


    false


  124. 生成的operator是: ReduceSinkOperator(map阶段运行)+GroupbyOperator（reduce阶段运行）








  132. 第一种情况：


  134. set hive.map.aggr=

    false


    ;


  136. set hive.groupby.skewindata=

    false


    ;


  138. SemanticAnalyzer.genGroupByPlan1MR(){


  140. （

    1


    ）ReduceSinkOperator： It will put all Group By keys and the

  141. distinct field (

    if


    any) in the map-reduce sort key, and all other fields

  142. in the map-reduce value.


  144. （

    2


    ）GroupbyOperator：GroupByDesc.Mode.COMPLETE，Reducer: iterate/merge (mode = COMPLETE)


  146. }




  150. 第二种情况：


  152. set hive.map.aggr=

    true


    ;


  154. set hive.groupby.skewindata=

    false


    ;


  156. SemanticAnalyzer.genGroupByPlanMapAggr1MR(){


  158. （

    1


    ）GroupByOperator：GroupByDesc.Mode.HASH，The agggregation

  159. evaluation functions are as follows: Mapper: iterate/terminatePartial

  160. (mode = HASH)


  162. （

    2


    ）ReduceSinkOperator：Partitioning Key: grouping key。Sorting Key:

  163. grouping key

    if


    no DISTINCT grouping + distinct key


    if


    DISTINCT


  165. （

    3


    ）GroupByOperator：GroupByDesc.Mode.MERGEPARTIAL，Reducer:

  166. iterate/terminate

    if


    DISTINCT merge/terminate


    if


    NO DISTINCT (mode =

  167. MERGEPARTIAL)


  169. }




  173. 第三种情况：


  175. set hive.map.aggr=

    false


    ;


  177. set hive.groupby.skewindata=

    true


    ;


  179. SemanticAnalyzer.genGroupByPlan2MR(){


  181. （

    1


    ）ReduceSinkOperator：Partitioning Key: random()


    if


    no DISTINCT

  182. grouping + distinct key

    if


    DISTINCT。Sorting Key: grouping key


    if


    no

  183. DISTINCT grouping + distinct key

    if


    DISTINCT


  185. （

    2


    ）GroupbyOperator：GroupByDesc.Mode.PARTIAL1，Reducer: iterate/terminatePartial (mode = PARTIAL1)


  187. （

    3


    ）ReduceSinkOperator：Partitioning Key: grouping key。Sorting

  188. Key: grouping key

    if


    no DISTINCT grouping + distinct key


    if


    DISTINCT


  190. （

    4


    ）GroupByOperator：GroupByDesc.Mode.FINAL，Reducer: merge/terminate (mode = FINAL)


  192. }




  196. 第四种情况：


  198. set hive.map.aggr=

    true


    ;


  200. set hive.groupby.skewindata=

    true


    ;


  202. SemanticAnalyzer.genGroupByPlanMapAggr2MR(){


  204. （

    1


    ）GroupbyOperator：GroupByDesc.Mode.HASH，Mapper: iterate/terminatePartial (mode = HASH)


  206. （

    2


    ）ReduceSinkOperator： Partitioning Key: random()


    if


    no

  207. DISTINCT grouping + distinct key

    if


    DISTINCT。 Sorting Key: grouping key


  208. if


    no DISTINCT grouping + distinct key


    if


    DISTINCT。


  210. （

    3


    ）GroupbyOperator：GroupByDesc.Mode.PARTIALS， Reducer:

  211. iterate/terminatePartial

    if


    DISTINCT merge/terminatePartial


    if


    NO

  212. DISTINCT (mode = MERGEPARTIAL)


  214. （

    4


    ）ReduceSinkOperator：Partitioining Key: grouping key。Sorting

  215. Key: grouping key

    if


    no DISTINCT grouping + distinct key


    if


    DISTINCT


  217. （

    5


    ）GroupByOperator：GroupByDesc.Mode.FINAL，Reducer: merge/terminate (mode = FINAL)


  219. }






  225. ReduceSinkOperator的processOp(Object row,

    int

  226. tag)会根据相应的条件设置Key的hash值，如第四种情况的第一个ReduceSinkOperator：Partitioning Key:

  227. random()

    if


    no DISTINCT grouping + distinct key


    if

  228. DISTINCT，如果没有DISTINCT字段，那么在OutputCollector.collect前会设置当前Key的hash值为一个随机

  229. 数，random =

    new


    Random(


    12345


    );。如果有DISTINCT字段，那么key的hash值跟grouping +

  230. distinct key有关。








  238. GroupByOperator：


  240. initializeOp(Configuration hconf)


  242. processOp(Object row,

    int


    tag)


  244. closeOp(

    boolean


    abort)


  246. forward(ArrayList<Object> keys, AggregationBuffer[] aggs)






  252. groupby10.q   groupby11.q


  254. set hive.map.aggr=

    false


    ;


  256. set hive.groupby.skewindata=

    false


    ;




  260. EXPLAIN


  262. FROM INPUT


  264. INSERT OVERWRITE TABLE dest1 SELECT INPUT.key,

  265. count(substr(INPUT.value,

    5


    )), count(distinct substr(INPUT.value,


    5


    ))

  266. GROUP BY INPUT.key;




  270. STAGE DEPENDENCIES:


  272. Stage-

    1


    is a root stage


  274. Stage-

    0


    depends on stages: Stage-


    1




  278. STAGE PLANS:


  280. Stage: Stage-

    1


  282. Map Reduce


  284. Alias -> Map Operator Tree:


  286. input


  288. TableScan


  290. alias: input


  292. Select Operator

    // insertSelectAllPlanForGroupBy


  294. expressions:


  296. expr: key


  298. type:

    int


  300. expr: value


  302. type: string


  304. outputColumnNames: key, value


  306. Reduce Output Operator


  308. key expressions:


  310. expr: key


  312. type:

    int


  314. expr: substr(value,

    5


    )


  316. type: string


  318. sort order: ++


  320. Map-reduce partition columns:


  322. expr: key


  324. type:

    int


  326. tag: –

    1


  328. Reduce Operator Tree:


  330. Group By Operator


  332. aggregations:


  334. expr: count(KEY._col1:

    0


    ._col0)


  336. expr: count(DISTINCT KEY._col1:

    0


    ._col0)


  338. bucketGroup:

    false


  340. keys:


  342. expr: KEY._col0


  344. type:

    int


  346. mode: complete


  348. outputColumnNames: _col0, _col1, _col2


  350. Select Operator


  352. expressions:


  354. expr: _col0


  356. type:

    int


  358. expr: _col1


  360. type: bigint


  362. expr: _col2


  364. type: bigint


  366. outputColumnNames: _col0, _col1, _col2


  368. Select Operator


  370. expressions:


  372. expr: _col0


  374. type:

    int


  376. expr: UDFToInteger(_col1)


  378. type:

    int


  380. expr: UDFToInteger(_col2)


  382. type:

    int


  384. outputColumnNames: _col0, _col1, _col2


  386. File Output Operator


  388. compressed:

    false


  390. GlobalTableId:

    1


  392. table:


  394. input format: org.apache.hadoop.mapred.TextInputFormat


  396. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  398. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  400. name: dest1




  404. Stage: Stage-

    0


  406. Move Operator


  408. tables:


  410. replace:

    true


  412. table:


  414. input format: org.apache.hadoop.mapred.TextInputFormat


  416. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  418. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  420. name: dest1










  430. set hive.map.aggr=

    true


    ;


  432. set hive.groupby.skewindata=

    false


    ;




  436. STAGE DEPENDENCIES:


  438. Stage-

    1


    is a root stage


  440. Stage-

    0


    depends on stages: Stage-


    1




  444. STAGE PLANS:


  446. Stage: Stage-

    1


  448. Map Reduce


  450. Alias -> Map Operator Tree:


  452. input


  454. TableScan


  456. alias: input


  458. Select Operator


  460. expressions:


  462. expr: key


  464. type:

    int


  466. expr: value


  468. type: string


  470. outputColumnNames: key, value


  472. Group By Operator


  474. aggregations:


  476. expr: count(substr(value,

    5


    ))


  478. expr: count(DISTINCT substr(value,

    5


    ))


  480. bucketGroup:

    false


  482. keys:


  484. expr: key


  486. type:

    int


  488. expr: substr(value,

    5


    )


  490. type: string


  492. mode: hash


  494. outputColumnNames: _col0, _col1, _col2, _col3


  496. Reduce Output Operator


  498. key expressions:


  500. expr: _col0


  502. type:

    int


  504. expr: _col1


  506. type: string


  508. sort order: ++


  510. Map-reduce partition columns:


  512. expr: _col0


  514. type:

    int


  516. tag: –

    1


  518. value expressions:


  520. expr: _col2


  522. type: bigint


  524. expr: _col3


  526. type: bigint


  528. Reduce Operator Tree:


  530. Group By Operator


  532. aggregations:


  534. expr: count(VALUE._col0)


  536. expr: count(DISTINCT KEY._col1:

    0


    ._col0)


  538. bucketGroup:

    false


  540. keys:


  542. expr: KEY._col0


  544. type:

    int


  546. mode: mergepartial


  548. outputColumnNames: _col0, _col1, _col2


  550. Select Operator


  552. expressions:


  554. expr: _col0


  556. type:

    int


  558. expr: _col1


  560. type: bigint


  562. expr: _col2


  564. type: bigint


  566. outputColumnNames: _col0, _col1, _col2


  568. Select Operator


  570. expressions:


  572. expr: _col0


  574. type:

    int


  576. expr: UDFToInteger(_col1)


  578. type:

    int


  580. expr: UDFToInteger(_col2)


  582. type:

    int


  584. outputColumnNames: _col0, _col1, _col2


  586. File Output Operator


  588. compressed:

    false


  590. GlobalTableId:

    1


  592. table:


  594. input format: org.apache.hadoop.mapred.TextInputFormat


  596. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  598. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  600. name: dest1




  604. Stage: Stage-

    0


  606. Move Operator


  608. tables:


  610. replace:

    true


  612. table:


  614. input format: org.apache.hadoop.mapred.TextInputFormat


  616. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  618. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  620. name: dest1


















  638. set hive.map.aggr=

    false


    ;


  640. set hive.groupby.skewindata=

    true


    ;




  644. STAGE DEPENDENCIES:


  646. Stage-

    1


    is a root stage


  648. Stage-

    2


    depends on stages: Stage-


    1


  650. Stage-

    0


    depends on stages: Stage-


    2




  654. STAGE PLANS:


  656. Stage: Stage-

    1


  658. Map Reduce


  660. Alias -> Map Operator Tree:


  662. input


  664. TableScan


  666. alias: input


  668. Select Operator


  670. expressions:


  672. expr: key


  674. type:

    int


  676. expr: value


  678. type: string


  680. outputColumnNames: key, value


  682. Reduce Output Operator


  684. key expressions:


  686. expr: key


  688. type:

    int


  690. expr: substr(value,

    5


    )


  692. type: string


  694. sort order: ++


  696. Map-reduce partition columns:


  698. expr: key


  700. type:

    int


  702. tag: –

    1


  704. Reduce Operator Tree:


  706. Group By Operator


  708. aggregations:


  710. expr: count(KEY._col1:

    0


    ._col0)


  712. expr: count(DISTINCT KEY._col1:

    0


    ._col0)


  714. bucketGroup:

    false


  716. keys:


  718. expr: KEY._col0


  720. type:

    int


  722. mode: partial1


  724. outputColumnNames: _col0, _col1, _col2


  726. File Output Operator


  728. compressed:

    false


  730. GlobalTableId:

    0


  732. table:


  734. input format: org.apache.hadoop.mapred.SequenceFileInputFormat


  736. output format: org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat




  740. Stage: Stage-

    2


  742. Map Reduce


  744. Alias -> Map Operator Tree:


  746. hdfs:

    //localhost:54310/tmp/hive-tianzhao/hive_2011-07-15_21-48-26_387_7978992474997402829/-mr-10002


  748. Reduce Output Operator


  750. key expressions:


  752. expr: _col0


  754. type:

    int


  756. sort order: +


  758. Map-reduce partition columns:


  760. expr: _col0


  762. type:

    int


  764. tag: –

    1


  766. value expressions:


  768. expr: _col1


  770. type: bigint


  772. expr: _col2


  774. type: bigint


  776. Reduce Operator Tree:


  778. Group By Operator


  780. aggregations:


  782. expr: count(VALUE._col0)


  784. expr: count(VALUE._col1)


  786. bucketGroup:

    false


  788. keys:


  790. expr: KEY._col0


  792. type:

    int


  794. mode:

    final


  796. outputColumnNames: _col0, _col1, _col2


  798. Select Operator


  800. expressions:


  802. expr: _col0


  804. type:

    int


  806. expr: _col1


  808. type: bigint


  810. expr: _col2


  812. type: bigint


  814. outputColumnNames: _col0, _col1, _col2


  816. Select Operator


  818. expressions:


  820. expr: _col0


  822. type:

    int


  824. expr: UDFToInteger(_col1)


  826. type:

    int


  828. expr: UDFToInteger(_col2)


  830. type:

    int


  832. outputColumnNames: _col0, _col1, _col2


  834. File Output Operator


  836. compressed:

    false


  838. GlobalTableId:

    1


  840. table:


  842. input format: org.apache.hadoop.mapred.TextInputFormat


  844. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  846. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  848. name: dest1




  852. Stage: Stage-

    0


  854. Move Operator


  856. tables:


  858. replace:

    true


  860. table:


  862. input format: org.apache.hadoop.mapred.TextInputFormat


  864. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  866. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  868. name: dest1










  878. set hive.map.aggr=

    true


    ;


  880. set hive.groupby.skewindata=

    true


    ;




  884. STAGE DEPENDENCIES:


  886. Stage-

    1


    is a root stage


  888. Stage-

    2


    depends on stages: Stage-


    1


  890. Stage-

    0


    depends on stages: Stage-


    2




  894. STAGE PLANS:


  896. Stage: Stage-

    1


  898. Map Reduce


  900. Alias -> Map Operator Tree:


  902. input


  904. TableScan


  906. alias: input


  908. Select Operator


  910. expressions:


  912. expr: key


  914. type:

    int


  916. expr: value


  918. type: string


  920. outputColumnNames: key, value


  922. Group By Operator


  924. aggregations:


  926. expr: count(substr(value,

    5


    ))


  928. expr: count(DISTINCT substr(value,

    5


    ))


  930. bucketGroup:

    false


  932. keys:


  934. expr: key


  936. type:

    int


  938. expr: substr(value,

    5


    )


  940. type: string


  942. mode: hash


  944. outputColumnNames: _col0, _col1, _col2, _col3


  946. Reduce Output Operator


  948. key expressions:


  950. expr: _col0


  952. type:

    int


  954. expr: _col1


  956. type: string


  958. sort order: ++


  960. Map-reduce partition columns:


  962. expr: _col0


  964. type:

    int


  966. tag: –

    1


  968. value expressions:


  970. expr: _col2


  972. type: bigint


  974. expr: _col3


  976. type: bigint


  978. Reduce Operator Tree:


  980. Group By Operator


  982. aggregations:


  984. expr: count(VALUE._col0)


  986. expr: count(DISTINCT KEY._col1:

    0


    ._col0)


  988. bucketGroup:

    false


  990. keys:


  992. expr: KEY._col0


  994. type:

    int


  996. mode: partials


  998. outputColumnNames: _col0, _col1, _col2


  1000. File Output Operator


  1002. compressed:

    false


  1004. GlobalTableId:

    0


  1006. table:


  1008. input format: org.apache.hadoop.mapred.SequenceFileInputFormat


  1010. output format: org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat




  1014. Stage: Stage-

    2


  1016. Map Reduce


  1018. Alias -> Map Operator Tree:


  1020. hdfs:

    //localhost:54310/tmp/hive-tianzhao/hive_2011-07-15_21-49-25_899_4946067838822964610/-mr-10002


  1022. Reduce Output Operator


  1024. key expressions:


  1026. expr: _col0


  1028. type:

    int


  1030. sort order: +


  1032. Map-reduce partition columns:


  1034. expr: _col0


  1036. type:

    int


  1038. tag: –

    1


  1040. value expressions:


  1042. expr: _col1


  1044. type: bigint


  1046. expr: _col2


  1048. type: bigint


  1050. Reduce Operator Tree:


  1052. Group By Operator


  1054. aggregations:


  1056. expr: count(VALUE._col0)


  1058. expr: count(VALUE._col1)


  1060. bucketGroup:

    false


  1062. keys:


  1064. expr: KEY._col0


  1066. type:

    int


  1068. mode:

    final


  1070. outputColumnNames: _col0, _col1, _col2


  1072. Select Operator


  1074. expressions:


  1076. expr: _col0


  1078. type:

    int


  1080. expr: _col1


  1082. type: bigint


  1084. expr: _col2


  1086. type: bigint


  1088. outputColumnNames: _col0, _col1, _col2


  1090. Select Operator


  1092. expressions:


  1094. expr: _col0


  1096. type:

    int


  1098. expr: UDFToInteger(_col1)


  1100. type:

    int


  1102. expr: UDFToInteger(_col2)


  1104. type:

    int


  1106. outputColumnNames: _col0, _col1, _col2


  1108. File Output Operator


  1110. compressed:

    false


  1112. GlobalTableId:

    1


  1114. table:


  1116. input format: org.apache.hadoop.mapred.TextInputFormat


  1118. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1120. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1122. name: dest1




  1126. Stage: Stage-

    0


  1128. Move Operator


  1130. tables:


  1132. replace:

    true


  1134. table:


  1136. input format: org.apache.hadoop.mapred.TextInputFormat


  1138. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1140. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1142. name: dest1


















  1160. set hive.map.aggr=

    false


    ;


  1162. set hive.groupby.skewindata=

    false


    ;




  1166. EXPLAIN extended


  1168. FROM INPUT


  1170. INSERT OVERWRITE TABLE dest1 SELECT INPUT.key,

  1171. count(substr(INPUT.value,

    5


    )), count(distinct substr(INPUT.value,


    5


    ))

  1172. GROUP BY INPUT.key;




  1176. STAGE DEPENDENCIES:


  1178. Stage-

    1


    is a root stage


  1180. Stage-

    0


    depends on stages: Stage-


    1




  1184. STAGE PLANS:


  1186. Stage: Stage-

    1


  1188. Map Reduce


  1190. Alias -> Map Operator Tree:


  1192. input


  1194. TableScan


  1196. alias: input


  1198. Select Operator


  1200. expressions:


  1202. expr: key


  1204. type:

    int


  1206. expr: value


  1208. type: string


  1210. outputColumnNames: key, value


  1212. Reduce Output Operator


  1214. key expressions:


  1216. expr: key


  1218. type:

    int


  1220. expr: substr(value,

    5


    )


  1222. type: string


  1224. sort order: ++


  1226. Map-reduce partition columns:


  1228. expr: key


  1230. type:

    int


  1232. tag: –

    1


  1234. Needs Tagging:

    false


  1236. Path -> Alias:


  1238. hdfs:

    //localhost:54310/user/hive/warehouse/input [input]


  1240. Path -> Partition:


  1242. hdfs:

    //localhost:54310/user/hive/warehouse/input


  1244. Partition


  1246. base file name: input


  1248. input format: org.apache.hadoop.mapred.TextInputFormat


  1250. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1252. properties:


  1254. bucket_count –

    1


  1256. columns key,value


  1258. columns.types

    int


    :string


  1260. file.inputformat org.apache.hadoop.mapred.TextInputFormat


  1262. file.outputformat org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1264. location hdfs:

    //localhost:54310/user/hive/warehouse/input


  1266. name input


  1268. serialization.ddl struct input { i32 key, string value}


  1270. serialization.format

    1


  1272. serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1274. transient_lastDdlTime

    1310523947


  1276. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe




  1280. input format: org.apache.hadoop.mapred.TextInputFormat


  1282. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1284. properties:


  1286. bucket_count –

    1


  1288. columns key,value


  1290. columns.types

    int


    :string


  1292. file.inputformat org.apache.hadoop.mapred.TextInputFormat


  1294. file.outputformat org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1296. location hdfs:

    //localhost:54310/user/hive/warehouse/input


  1298. name input


  1300. serialization.ddl struct input { i32 key, string value}


  1302. serialization.format

    1


  1304. serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1306. transient_lastDdlTime

    1310523947


  1308. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1310. name: input


  1312. name: input


  1314. Reduce Operator Tree:


  1316. Group By Operator


  1318. aggregations:


  1320. expr: count(KEY._col1:

    0


    ._col0)


  1322. expr: count(DISTINCT KEY._col1:

    0


    ._col0)


  1324. bucketGroup:

    false


  1326. keys:


  1328. expr: KEY._col0


  1330. type:

    int


  1332. mode: complete


  1334. outputColumnNames: _col0, _col1, _col2


  1336. Select Operator


  1338. expressions:


  1340. expr: _col0


  1342. type:

    int


  1344. expr: _col1


  1346. type: bigint


  1348. expr: _col2


  1350. type: bigint


  1352. outputColumnNames: _col0, _col1, _col2


  1354. Select Operator


  1356. expressions:


  1358. expr: _col0


  1360. type:

    int


  1362. expr: UDFToInteger(_col1)


  1364. type:

    int


  1366. expr: UDFToInteger(_col2)


  1368. type:

    int


  1370. outputColumnNames: _col0, _col1, _col2


  1372. File Output Operator


  1374. compressed:

    false


  1376. GlobalTableId:

    1


  1378. directory: hdfs:

    //localhost:54310/tmp/hive-tianzhao/hive_2011-07-15_21-50-38_510_6852880850328147221/-ext-10000


  1380. NumFilesPerFileSink:

    1


  1382. table:


  1384. input format: org.apache.hadoop.mapred.TextInputFormat


  1386. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1388. properties:


  1390. bucket_count –

    1


  1392. columns key,val1,val2


  1394. columns.types

    int


    :


    int


    :


    int


  1396. file.inputformat org.apache.hadoop.mapred.TextInputFormat


  1398. file.outputformat org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1400. location hdfs:

    //localhost:54310/user/hive/warehouse/dest1


  1402. name dest1


  1404. serialization.ddl struct dest1 { i32 key, i32 val1, i32 val2}


  1406. serialization.format

    1


  1408. serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1410. transient_lastDdlTime

    1310523946


  1412. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1414. name: dest1


  1416. TotalFiles:

    1


  1418. MultiFileSpray:

    false




  1422. Stage: Stage-

    0


  1424. Move Operator


  1426. tables:


  1428. replace:

    true


  1430. source: hdfs:

    //localhost:54310/tmp/hive-tianzhao/hive_2011-07-15_21-50-38_510_6852880850328147221/-ext-10000


  1432. table:


  1434. input format: org.apache.hadoop.mapred.TextInputFormat


  1436. output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1438. properties:


  1440. bucket_count –

    1


  1442. columns key,val1,val2


  1444. columns.types

    int


    :


    int


    :


    int


  1446. file.inputformat org.apache.hadoop.mapred.TextInputFormat


  1448. file.outputformat org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat


  1450. location hdfs:

    //localhost:54310/user/hive/warehouse/dest1


  1452. name dest1


  1454. serialization.ddl struct dest1 { i32 key, i32 val1, i32 val2}


  1456. serialization.format

    1


  1458. serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1460. transient_lastDdlTime

    1310523946


  1462. serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe


  1464. name: dest1


  1466. tmp directory: hdfs:

    //localhost:54310/tmp/hive-tianzhao/hive_2011-07-15_21-50-38_510_6852880850328147221/-ext-10001








  1474. ABSTRACT SYNTAX TREE:


  1476. (TOK_QUERY


  1478. (TOK_FROM (TOK_TABREF INPUT))


  1480. (TOK_INSERT


  1482. (TOK_DESTINATION (TOK_TAB dest1))


  1484. (TOK_SELECT


  1486. (TOK_SELEXPR (. (TOK_TABLE_OR_COL INPUT) key))


  1488. (TOK_SELEXPR (TOK_FUNCTION count (TOK_FUNCTION substr (. (TOK_TABLE_OR_COL INPUT) value)

    5


    )))


  1490. (TOK_SELEXPR (TOK_FUNCTIONDI count (TOK_FUNCTION substr (. (TOK_TABLE_OR_COL INPUT) value)

    5


    )))


  1492. )


  1494. (TOK_GROUPBY (. (TOK_TABLE_OR_COL INPUT) key))


  1496. )


  1498. )








  1506. SemanticAnalyzer.genBodyPlan(QB qb, Operator input){



  1508. if


    (qbp.getAggregationExprsForClause(dest).size() !=


    0


  1510. || getGroupByForClause(qbp, dest).size() >

    0


    ) {


    //如果有聚合函数或者有groupby，则执行下面的操作



  1512. //multiple distincts is not supported with skew in data



  1514. if


    (conf.getVar(HiveConf.ConfVars.HIVEGROUPBYSKEW)


  1516. .equalsIgnoreCase(

    “true”


    ) &&


  1518. qbp.getDistinctFuncExprsForClause(dest).size() >

    1


    ) {



  1520. throw




    new


    SemanticException(ErrorMsg.UNSUPPORTED_MULTIPLE_DISTINCTS.


  1522. getMsg());


  1524. }



  1526. // insert a select operator here used by the ColumnPruner to reduce



  1528. // the data to shuffle


  1530. curr = insertSelectAllPlanForGroupBy(dest, curr);

    //生成一个SelectOperator，所有的字段都会选取，selectStar=true。



  1532. if


    (conf.getVar(HiveConf.ConfVars.HIVEMAPSIDEAGGREGATE)


  1534. .equalsIgnoreCase(

    “true”


    )) {



  1536. if


    (conf.getVar(HiveConf.ConfVars.HIVEGROUPBYSKEW)


  1538. .equalsIgnoreCase(

    “false”


    )) {


  1540. curr = genGroupByPlanMapAggr1MR(dest, qb, curr);


  1542. }

    else


    {


  1544. curr = genGroupByPlanMapAggr2MR(dest, qb, curr);


  1546. }


  1548. }

    else




    if


    (conf.getVar(HiveConf.ConfVars.HIVEGROUPBYSKEW)


  1550. .equalsIgnoreCase(

    “true”


    )) {


  1552. curr = genGroupByPlan2MR(dest, qb, curr);


  1554. }

    else


    {


  1556. curr = genGroupByPlan1MR(dest, qb, curr);


  1558. }


  1560. }


  1562. }




  1566. distince：


  1568. count.q.out


  1570. groupby11.q.out


  1572. groupby10.q.out


  1574. nullgroup4_multi_distinct.q.out


  1576. join18.q.out


  1578. groupby_bigdata.q.out


  1580. join18_multi_distinct.q.out


  1582. nullgroup4.q.out


  1584. auto_join18_multi_distinct.q.out


  1586. auto_join18.q.out




  1590. （

    1


    ）map端部分聚合，数据无倾斜，一个MR生成。


  1592. genGroupByPlanMapAggr1MR，生成三个Operator：


  1594. （

    1.1


    ）GroupByOperator：map-side partial aggregation，由genGroupByPlanMapGroupByOperator方法生成：


  1596. 处理groupby子句，getGroupByForClause，groupby的column加入groupByKeys和outputColumnNames


  1598. 处理select中的Distinct，getDistinctFuncExprsForClause，Distinct的column，加入groupByKeys和outputColumnNames


  1600. 处理聚合函数，getAggregationExprsForClause，生成AggregationDesc加入aggregations，生成column加入outputColumnNames



  1602. public


    GroupByDesc(



  1604. final


    Mode mode,



  1606. final


    java.util.ArrayList<java.lang.String> outputColumnNames,



  1608. final


    java.util.ArrayList<ExprNodeDesc> keys,



  1610. final


    java.util.ArrayList<org.apache.hadoop.hive.ql.plan.AggregationDesc> aggregators,



  1612. final




    boolean


    groupKeyNotReductionKey,


    float


    groupByMemoryUsage,


    float


    memoryThreshold) {



  1614. this


    (mode, outputColumnNames, keys, aggregators, groupKeyNotReductionKey,



  1616. false


    , groupByMemoryUsage, memoryThreshold);


  1618. }


  1620. mode：GroupByDesc.Mode.HASH


  1622. outputColumnNames：groupby+Distinct+Aggregation


  1624. keys：groupby+Distinct


  1626. aggregators：Aggregation


  1628. groupKeyNotReductionKey：

    false


  1630. groupByMemoryUsage：默认为

    0.5


  1632. memoryThreshold：默认为

    0.9




  1636. （

    1.2


    ）ReduceSinkOperator


  1638. 处理groupby子句，getGroupByForClause，groupby的column加入reduceKeys和outputKeyColumnNames


  1640. 处理select中的Distinct，getDistinctFuncExprsForClause，Distinct的column，加入reduceKeys和outputKeyColumnNames


  1642. 处理聚合函数，getAggregationExprsForClause，需要做聚合的column加入reduceValues和outputValueColumnNames



  1644. public


    ReduceSinkDesc(java.util.ArrayList<ExprNodeDesc> keyCols,



  1646. int


    numDistributionKeys,


  1648. java.util.ArrayList<ExprNodeDesc> valueCols,


  1650. java.util.ArrayList<java.lang.String> outputKeyColumnNames,


  1652. List<List<Integer>> distinctColumnIndices,


  1654. java.util.ArrayList<java.lang.String> outputValueColumnNames,

    int


    tag,


  1656. java.util.ArrayList<ExprNodeDesc> partitionCols,

    int


    numReducers,



  1658. final


    TableDesc keySerializeInfo,


    final


    TableDesc valueSerializeInfo) {



  1660. this


    .keyCols = keyCols;


    // 为reduceKeys，groupby+distinct



  1662. this


    .numDistributionKeys = numDistributionKeys;


    // grpByExprs.size()



  1664. this


    .valueCols = valueCols;


    //reduceValues，聚合函数



  1666. this


    .outputKeyColumnNames = outputKeyColumnNames;


    //outputKeyColumnNames



  1668. this


    .outputValueColumnNames = outputValueColumnNames;


    //outputValueColumnNames



  1670. this


    .tag = tag;


    // -1



  1672. this


    .numReducers = numReducers;


    // 一般都是-1



  1674. this


    .partitionCols = partitionCols;


    // groupby



  1676. this


    .keySerializeInfo = keySerializeInfo;



  1678. this


    .valueSerializeInfo = valueSerializeInfo;



  1680. this


    .distinctColumnIndices = distinctColumnIndices;


  1682. }




  1686. （

    1.3


    ）GroupByOperator


  1688. 处理groupby子句，getGroupByForClause，groupby的column加入reduceKeys和outputKeyColumnNames


  1690. 处理聚合函数，getAggregationExprsForClause，需要做聚合的column加入reduceValues和outputValueColumnNames



  1692. public


    GroupByDesc(



  1694. final


    Mode mode,



  1696. final


    java.util.ArrayList<java.lang.String> outputColumnNames,



  1698. final


    java.util.ArrayList<ExprNodeDesc> keys,



  1700. final


    java.util.ArrayList<org.apache.hadoop.hive.ql.plan.AggregationDesc> aggregators,



  1702. final




    boolean


    groupKeyNotReductionKey,


    float


    groupByMemoryUsage,


    float


    memoryThreshold) {



  1704. this


    (mode, outputColumnNames, keys, aggregators, groupKeyNotReductionKey,



  1706. false


    , groupByMemoryUsage, memoryThreshold);


  1708. }


  1710. mode：GroupByDesc.Mode.MERGEPARTIAL


  1712. outputColumnNames：groupby+Aggregation


  1714. keys：groupby


  1716. aggregators：Aggregation


  1718. groupKeyNotReductionKey：

    false


  1720. groupByMemoryUsage：默认为

    0.5


  1722. memoryThreshold：默认为

    0.9




  1726. （

    2


    ）map端部分聚合，数据倾斜，两个MR生成。


  1728. genGroupByPlanMapAggr2MR：


  1730. （

    2.1


    ）GroupByOperator：map-side partial aggregation，由genGroupByPlanMapGroupByOperator方法生成：


  1732. 处理groupby子句，getGroupByForClause，groupby的column加入groupByKeys和outputColumnNames


  1734. 处理select中的Distinct，getDistinctFuncExprsForClause，Distinct的column，加入groupByKeys和outputColumnNames


  1736. 处理聚合函数，getAggregationExprsForClause，生成AggregationDesc加入aggregations，生成column加入outputColumnNames



  1738. public


    GroupByDesc(



  1740. final


    Mode mode,



  1742. final


    java.util.ArrayList<java.lang.String> outputColumnNames,



  1744. final


    java.util.ArrayList<ExprNodeDesc> keys,



  1746. final


    java.util.ArrayList<org.apache.hadoop.hive.ql.plan.AggregationDesc> aggregators,



  1748. final




    boolean


    groupKeyNotReductionKey,


    float


    groupByMemoryUsage,


    float


    memoryThreshold) {



  1750. this


    (mode, outputColumnNames, keys, aggregators, groupKeyNotReductionKey,



  1752. false


    , groupByMemoryUsage, memoryThreshold);


  1754. }


  1756. mode：GroupByDesc.Mode.HASH


  1758. outputColumnNames：groupby+Distinct+Aggregation


  1760. keys：groupby+Distinct


  1762. aggregators：Aggregation


  1764. groupKeyNotReductionKey：

    false


  1766. groupByMemoryUsage：默认为

    0.5


  1768. memoryThreshold：默认为

    0.9




  1772. （

    2.2


    ）ReduceSinkOperator


  1774. 处理groupby子句，getGroupByForClause，groupby的column加入reduceKeys和outputKeyColumnNames


  1776. 处理select中的Distinct，getDistinctFuncExprsForClause，Distinct的column，加入reduceKeys和outputKeyColumnNames


  1778. 处理聚合函数，getAggregationExprsForClause，需要做聚合的column加入reduceValues和outputValueColumnNames



  1780. public


    ReduceSinkDesc(java.util.ArrayList<ExprNodeDesc> keyCols,



  1782. int


    numDistributionKeys,


  1784. java.util.ArrayList<ExprNodeDesc> valueCols,


  1786. java.util.ArrayList<java.lang.String> outputKeyColumnNames,


  1788. List<List<Integer>> distinctColumnIndices,


  1790. java.util.ArrayList<java.lang.String> outputValueColumnNames,

    int


    tag,


  1792. java.util.ArrayList<ExprNodeDesc> partitionCols,

    int


    numReducers,



  1794. final


    TableDesc keySerializeInfo,


    final


    TableDesc valueSerializeInfo) {



  1796. this


    .keyCols = keyCols;


    // 为reduceKeys，groupby+distinct



  1798. this


    .numDistributionKeys = numDistributionKeys;


    // grpByExprs.size()



  1800. this


    .valueCols = valueCols;


    //reduceValues，聚合函数



  1802. this


    .outputKeyColumnNames = outputKeyColumnNames;


    //outputKeyColumnNames



  1804. this


    .outputValueColumnNames = outputValueColumnNames;


    //outputValueColumnNames



  1806. this


    .tag = tag;


    // -1



  1808. this


    .numReducers = numReducers;


    // 一般都是-1



  1810. this


    .partitionCols = partitionCols;


    // groupby



  1812. this


    .keySerializeInfo = keySerializeInfo;



  1814. this


    .valueSerializeInfo = valueSerializeInfo;



  1816. this


    .distinctColumnIndices = distinctColumnIndices;


  1818. }




  1822. （

    2.3


    ）GroupByOperator


  1824. 处理groupby子句，getGroupByForClause，groupby的column加入groupByKeys和outputColumnNames


  1826. 处理聚合函数，getAggregationExprsForClause，生成AggregationDesc加入aggregations，生成column加入outputColumnNames



  1828. public


    GroupByDesc(



  1830. final


    Mode mode,



  1832. final


    java.util.ArrayList<java.lang.String> outputColumnNames,



  1834. final


    java.util.ArrayList<ExprNodeDesc> keys,



  1836. final


    java.util.ArrayList<org.apache.hadoop.hive.ql.plan.AggregationDesc> aggregators,



  1838. final




    boolean


    groupKeyNotReductionKey,


    float


    groupByMemoryUsage,


    float


    memoryThreshold) {



  1840. this


    (mode, outputColumnNames, keys, aggregators, groupKeyNotReductionKey,



  1842. false


    , groupByMemoryUsage, memoryThreshold);


  1844. }


  1846. mode：GroupByDesc.Mode.PARTIALS


  1848. outputColumnNames：groupby+Aggregation


  1850. keys：groupby


  1852. aggregators：Aggregation


  1854. groupKeyNotReductionKey：

    false


  1856. groupByMemoryUsage：默认为

    0.5


  1858. memoryThreshold：默认为

    0.9




  1862. （

    2.4


    ）ReduceSinkOperator


  1864. 处理groupby子句，getGroupByForClause，groupby的column加入reduceKeys和outputColumnNames


  1866. 处理聚合函数，getAggregationExprsForClause，需要做聚合的column加入reduceValues和outputColumnNames



  1868. public


    ReduceSinkDesc(java.util.ArrayList<ExprNodeDesc> keyCols,



  1870. int


    numDistributionKeys,


  1872. java.util.ArrayList<ExprNodeDesc> valueCols,


  1874. java.util.ArrayList<java.lang.String> outputKeyColumnNames,


  1876. List<List<Integer>> distinctColumnIndices,


  1878. java.util.ArrayList<java.lang.String> outputValueColumnNames,

    int


    tag,


  1880. java.util.ArrayList<ExprNodeDesc> partitionCols,

    int


    numReducers,



  1882. final


    TableDesc keySerializeInfo,


    final


    TableDesc valueSerializeInfo) {



  1884. this


    .keyCols = keyCols;


    // 为reduceKeys，groupby



  1886. this


    .numDistributionKeys = numDistributionKeys;


    // grpByExprs.size()



  1888. this


    .valueCols = valueCols;


    //reduceValues，聚合函数



  1890. this


    .outputKeyColumnNames = outputKeyColumnNames;


    //outputKeyColumnNames



  1892. this


    .outputValueColumnNames = outputValueColumnNames;


    //outputValueColumnNames



  1894. this


    .tag = tag;


    // -1



  1896. this


    .numReducers = numReducers;


    // 一般都是-1



  1898. this


    .partitionCols = partitionCols;


    // groupby



  1900. this


    .keySerializeInfo = keySerializeInfo;



  1902. this


    .valueSerializeInfo = valueSerializeInfo;



  1904. this


    .distinctColumnIndices = distinctColumnIndices;


  1906. }




  1910. （

    2.5


    ）GroupByOperator


  1912. 处理groupby子句，getGroupByForClause，groupby的column加入groupByKeys和outputColumnNames


  1914. 处理聚合函数，getAggregationExprsForClause，生成AggregationDesc加入aggregations，需要做聚合的column加入outputColumnNames



  1916. public


    GroupByDesc(



  1918. final


    Mode mode,



  1920. final


    java.util.ArrayList<java.lang.String> outputColumnNames,



  1922. final


    java.util.ArrayList<ExprNodeDesc> keys,



  1924. final


    java.util.ArrayList<org.apache.hadoop.hive.ql.plan.AggregationDesc> aggregators,



  1926. final




    boolean


    groupKeyNotReductionKey,


    float


    groupByMemoryUsage,


    float


    memoryThreshold) {



  1928. this


    (mode, outputColumnNames, keys, aggregators, groupKeyNotReductionKey,



  1930. false


    , groupByMemoryUsage, memoryThreshold);


  1932. }


  1934. mode：GroupByDesc.Mode.FINAL


  1936. outputColumnNames：groupby+Aggregation


  1938. keys：groupby


  1940. aggregators：Aggregation


  1942. groupKeyNotReductionKey：

    false


  1944. groupByMemoryUsage：默认为

    0.5


  1946. memoryThreshold：默认为

    0.9






  1952. （

    3


    ）map端不部分聚合，数据倾斜，两个MR生成。


  1954. genGroupByPlan2MR：




  1958. （

    3.1


    ）ReduceSinkOperator


  1960. 处理groupby子句，getGroupByForClause，groupby的column加入reduceKeys和outputKeyColumnNames


  1962. 处理select中的Distinct，getDistinctFuncExprsForClause，Distinct的column，加入reduceKeys和outputKeyColumnNames


  1964. 处理聚合函数，getAggregationExprsForClause，需要做聚合的column加入reduceValues和outputValueColumnNames



  1966. public


    ReduceSinkDesc(java.util.ArrayList<ExprNodeDesc> keyCols,



  1968. int


    numDistributionKeys,


  1970. java.util.ArrayList<ExprNodeDesc> valueCols,


  1972. java.util.ArrayList<java.lang.String> outputKeyColumnNames,


  1974. List<List<Integer>> distinctColumnIndices,


  1976. java.util.ArrayList<java.lang.String> outputValueColumnNames,

    int


    tag,


  1978. java.util.ArrayList<ExprNodeDesc> partitionCols,

    int


    numReducers,



  1980. final


    TableDesc keySerializeInfo,


    final


    TableDesc valueSerializeInfo) {



  1982. this


    .keyCols = keyCols;


    // 为reduceKeys，groupby+distinct



  1984. this


    .numDistributionKeys = numDistributionKeys;


    // grpByExprs.size()



  1986. this


    .valueCols = valueCols;


    //reduceValues，聚合函数



  1988. this


    .outputKeyColumnNames = outputKeyColumnNames;


    //outputKeyColumnNames



  1990. this


    .outputValueColumnNames = outputValueColumnNames;


    //outputValueColumnNames



  1992. this


    .tag = tag;


    // -1



  1994. this


    .numReducers = numReducers;


    // 一般都是-1



  1996. this


    .partitionCols = partitionCols;


    // groupby



  1998. this


    .keySerializeInfo = keySerializeInfo;



  2000. this


    .valueSerializeInfo = valueSerializeInfo;



  2002. this


    .distinctColumnIndices = distinctColumnIndices;


  2004. }




  2008. （

    3.2


    ）GroupByOperator


  2010. 处理groupby子句，getGroupByForClause，groupby的column加入groupByKeys和outputColumnNames


  2012. 处理聚合函数，getAggregationExprsForClause，生成AggregationDesc加入aggregations，生成column加入outputColumnNames



  2014. public


    GroupByDesc(



  2016. final


    Mode mode,



  2018. final


    java.util.ArrayList<java.lang.String> outputColumnNames,



  2020. final


    java.util.ArrayList<ExprNodeDesc> keys,



  2022. final


    java.util.ArrayList<org.apache.hadoop.hive.ql.plan.AggregationDesc> aggregators,



  2024. final




    boolean


    groupKeyNotReductionKey,


    float


    groupByMemoryUsage,


    float


    memoryThreshold) {



  2026. this


    (mode, outputColumnNames, keys, aggregators, groupKeyNotReductionKey,



  2028. false


    , groupByMemoryUsage, memoryThreshold);


  2030. }


  2032. mode：GroupByDesc.Mode.PARTIAL1


  2034. outputColumnNames：groupby+Aggregation


  2036. keys：groupby


  2038. aggregators：Aggregation


  2040. groupKeyNotReductionKey：

    false


  2042. groupByMemoryUsage：默认为

    0.5


  2044. memoryThreshold：默认为

    0.9




  2048. （

    3.3


    ）ReduceSinkOperator


  2050. 处理groupby子句，getGroupByForClause，groupby的column加入reduceKeys和outputColumnNames


  2052. 处理聚合函数，getAggregationExprsForClause，需要做聚合的column加入reduceValues和outputColumnNames



  2054. public


    ReduceSinkDesc(java.util.ArrayList<ExprNodeDesc> keyCols,



  2056. int


    numDistributionKeys,


  2058. java.util.ArrayList<ExprNodeDesc> valueCols,


  2060. java.util.ArrayList<java.lang.String> outputKeyColumnNames,


  2062. List<List<Integer>> distinctColumnIndices,


  2064. java.util.ArrayList<java.lang.String> outputValueColumnNames,

    int


    tag,


  2066. java.util.ArrayList<ExprNodeDesc> partitionCols,

    int


    numReducers,



  2068. final


    TableDesc keySerializeInfo,


    final


    TableDesc valueSerializeInfo) {



  2070. this


    .keyCols = keyCols;


    // 为reduceKeys，groupby



  2072. this


    .numDistributionKeys = numDistributionKeys;


    // grpByExprs.size()



  2074. this


    .valueCols = valueCols;


    //reduceValues，聚合函数



  2076. this


    .outputKeyColumnNames = outputKeyColumnNames;


    //outputKeyColumnNames



  2078. this


    .outputValueColumnNames = outputValueColumnNames;


    //outputValueColumnNames



  2080. this


    .tag = tag;


    // -1



  2082. this


    .numReducers = numReducers;


    // 一般都是-1



  2084. this


    .partitionCols = partitionCols;


    // groupby



  2086. this


    .keySerializeInfo = keySerializeInfo;



  2088. this


    .valueSerializeInfo = valueSerializeInfo;



  2090. this


    .distinctColumnIndices = distinctColumnIndices;


  2092. }




  2096. （

    3.4


    ）GroupByOperator


  2098. 处理groupby子句，getGroupByForClause，groupby的column加入groupByKeys和outputColumnNames


  2100. 处理聚合函数，getAggregationExprsForClause，生成AggregationDesc加入aggregations，需要做聚合的column加入outputColumnNames



  2102. public


    GroupByDesc(



  2104. final


    Mode mode,



  2106. final


    java.util.ArrayList<java.lang.String> outputColumnNames,



  2108. final


    java.util.ArrayList<ExprNodeDesc> keys,



  2110. final


    java.util.ArrayList<org.apache.hadoop.hive.ql.plan.AggregationDesc> aggregators,



  2112. final




    boolean


    groupKeyNotReductionKey,


    float


    groupByMemoryUsage,


    float


    memoryThreshold) {



  2114. this


    (mode, outputColumnNames, keys, aggregators, groupKeyNotReductionKey,



  2116. false


    , groupByMemoryUsage, memoryThreshold);


  2118. }


  2120. mode：GroupByDesc.Mode.FINAL


  2122. outputColumnNames：groupby+Aggregation


  2124. keys：groupby


  2126. aggregators：Aggregation


  2128. groupKeyNotReductionKey：

    false


  2130. groupByMemoryUsage：默认为

    0.5


  2132. memoryThreshold：默认为

    0.9






  2138. （

    4


    ）map端不部分聚合，数据无倾斜，一个MR生成。


  2140. genGroupByPlan1MR：


  2142. （

    4.1


    ）ReduceSinkOperator


  2144. 处理groupby子句，getGroupByForClause，groupby的column加入reduceKeys和outputKeyColumnNames


  2146. 处理select中的Distinct，getDistinctFuncExprsForClause，Distinct的column，加入reduceKeys和outputKeyColumnNames


  2148. 处理聚合函数，getAggregationExprsForClause，需要做聚合的column加入reduceValues和outputValueColumnNames



  2150. public


    ReduceSinkDesc(java.util.ArrayList<ExprNodeDesc> keyCols,



  2152. int


    numDistributionKeys,


  2154. java.util.ArrayList<ExprNodeDesc> valueCols,


  2156. java.util.ArrayList<java.lang.String> outputKeyColumnNames,


  2158. List<List<Integer>> distinctColumnIndices,


  2160. java.util.ArrayList<java.lang.String> outputValueColumnNames,

    int


    tag,


  2162. java.util.ArrayList<ExprNodeDesc> partitionCols,

    int


    numReducers,



  2164. final


    TableDesc keySerializeInfo,


    final


    TableDesc valueSerializeInfo) {



  2166. this


    .keyCols = keyCols;


    // 为reduceKeys，groupby+distinct



  2168. this


    .numDistributionKeys = numDistributionKeys;


    // grpByExprs.size()



  2170. this


    .valueCols = valueCols;


    //reduceValues，聚合函数



  2172. this


    .outputKeyColumnNames = outputKeyColumnNames;


    //outputKeyColumnNames



  2174. this


    .outputValueColumnNames = outputValueColumnNames;


    //outputValueColumnNames



  2176. this


    .tag = tag;


    // -1



  2178. this


    .numReducers = numReducers;


    // 一般都是-1



  2180. this


    .partitionCols = partitionCols;


    // groupby



  2182. this


    .keySerializeInfo = keySerializeInfo;



  2184. this


    .valueSerializeInfo = valueSerializeInfo;



  2186. this


    .distinctColumnIndices = distinctColumnIndices;


  2188. }




  2192. （

    4.2


    ）GroupByOperator


  2194. 处理groupby子句，getGroupByForClause，groupby的column加入reduceKeys和outputKeyColumnNames


  2196. 处理聚合函数，getAggregationExprsForClause，需要做聚合的column加入reduceValues和outputValueColumnNames



  2198. public


    GroupByDesc(



  2200. final


    Mode mode,



  2202. final


    java.util.ArrayList<java.lang.String> outputColumnNames,



  2204. final


    java.util.ArrayList<ExprNodeDesc> keys,



  2206. final


    java.util.ArrayList<org.apache.hadoop.hive.ql.plan.AggregationDesc> aggregators,



  2208. final




    boolean


    groupKeyNotReductionKey,


    float


    groupByMemoryUsage,


    float


    memoryThreshold) {



  2210. this


    (mode, outputColumnNames, keys, aggregators, groupKeyNotReductionKey,



  2212. false


    , groupByMemoryUsage, memoryThreshold);


  2214. }


  2216. mode：GroupByDesc.Mode.COMPLETE


  2218. outputColumnNames：groupby+Aggregation


  2220. keys：groupby


  2222. aggregators：Aggregation


  2224. groupKeyNotReductionKey：

    false


  2226. groupByMemoryUsage：默认为

    0.5


  2228. memoryThreshold：默认为

    0.9








  2236. SemanticAnalyzer.genBodyPlan


  2238. optimizeMultiGroupBy  (multi-group by with the same distinct)


  2240. groupby10.q  groupby11.q