Hive 原理和调优_hive distribute by rand

hive原理

Hive是基于Hadoop的一个数据仓库工具，可以将结构化的数据文件映射为一张数据库表，并提供类SQL的查询功能。
其基本原理是将HQL语句自动转换成MapReduce任务。
Hive建立在Hadoop的其它组件之上:依赖于HDFS进行数据存储；依赖于MapReduce完成查询操作。

一、hive组件组成
1、Driver组件：该组件包括：Compiler、Optimizer、Executor,它可以将Hive的编译、解析、优化转化为MapReduce
任务提交给Hadoop1中的JobTracker或者是Hadoop2中的SourceManager来进行实际的执行相应的任务。
2. MetaStore组件：存储着hive的元数据信息，将自己的元数据存储到了关系型数据库当中，支持的数据库主要有：
Mysql、Derby、支持把metastore独立出来放在远程的集群上面，使得hive更加健壮。元数据主要包括了表的名称、
表的列、分区和属性、表的属性（是不是外部表等等）、表的数据所在的目录。

二、流程大致步骤为：

1. 用户提交查询等任务给Driver。
2. 编译器获得该用户的任务Plan。
3. 编译器Compiler根据用户任务去MetaStore中获取需要的Hive的元数据信息。
4. 编译器Compiler得到元数据信息，对任务进行编译，HiveQL->抽象语法树->查询块->逻辑查询计划->
   重写逻辑查询计划->物理查询计划(MapReduce)->选择最佳的策略。
5. 将最终的计划提交给Driver。
6. Driver将计划Plan转交给ExecutionEngine去执行，获取元数据信息，提交给JobTracker或者SourceManager执行该任务，
   任务会直接读取HDFS中文件进行相应的操作。
7. 获取执行的结果。
8. 取得并返回执行结果。

三、hive优缺点，应用场景
优点：
1、高可靠、高容错：HiveServer采用主备模式，双MetaStore，超时重试
2、类SQL：类似SQL语法，内置大量函数
3、可扩展：自定义存储格式，自定义函数（UDF/UDAF/UDTF)
4、多接口：Beeline，JDBC，Thrift，Python
缺点：
1、延迟较高：默认M/R为执行引擎，M/R启动有延迟
2、暂不支持事务：暂不支持事务类操作
3、不适用OLTP：暂不支持列级别的数据添加、更新、删除操作
4、暂不支持存储过程：目前还不能支持存储过程，只能通过UDF来实现一些逻辑处理
应用场景：
1、数据挖掘：用户行为分析，兴趣分区，区域展示
2、非实时分析：日志分析，文本分析
3、数据汇总：每天/每周用户点击数，流量统计
4、作为数据仓库：数据抽取，数据加载，数据转换

四、hive与传统数据库比较
1、存储：
hive: hdfs,理论上有无限拓展的可能.
传统：集群存储，存在容量上限，而且伴随容量的增长，计算速度急剧下降。只能适应于数据量比较小的商业应用，
对于超大规模数据无能为力。
2、执行引擎：
hive:依赖于MapReduce框架，可进行的各类优化较少，但是比较简单。
传统：可以选择更加高效的算法来执行查询，也可以进行更多的优化措施来提高速度。
3、使用方式：
hive：HQL（类似SQL）。
传统：SQL。
4、分析速度：
hive：计算依赖于MapReduce和集群规模，易拓展，在大数据量情况下，远远快于普通数据仓库。
传统：在数据容量较小时非常快速，数据量较大时，急剧下降。

五、托管表和外部表
Hive 默认创建Managed Table，由Hive来管理数据，意味着Hive会将数据移动到数据仓库目录。
另外一种选择是创建External Table，这时Hive会到仓库目录以外的位置访问数据。
建议：
如果所有处理都由Hive完成，建议使用Managed Table。
如果要用Hive和其它工具来处理同一个数据集，应该使用External Tables。
CREATE/LOAD：
1、托管表：把数据移到仓库目录
2、外部表：创建表时指明外部数据的位置
DROP：
1、托管表：元数据和数据会被一起删除
2、外部表：只删除元数据

六、分区和分桶
1、分区：数据表可以按照某个字段的值划分分区，在将数据加载到表内之前，需要明确指出加载的数据所属分区。
CREATE TABLE logs(timestamp BIGINT,line STRING)PARTITIONED BY (date STRING);
load data inpath ‘xxx’ into table logs partition(date=‘20150717’); //加载数据
alter table logs add partition(date=‘20160717’); //添加分区
alter table logs drop partition (date=‘20161118’); //删除分区
show partitions 表名; // 显示分区
注意：PARTITONED BY子句中定义的列是表中正式的列（分区列），但是数据文件内并不包含这些列。
(1)每个分区是一个目录
(2)分区数量不固定
(3)分区下可再有分区或者桶
(4)分区可以很明显的提高查询效率
2、桶：数据可以根据桶的方式将不同数据方式不同的桶中
SET hive.enforce.bucketing=true;
set mapreduce.job.reduces=分桶数;
(1)每个桶是一个文件
(2)建表时指定桶个数，桶内可排序
(3)数据按照某个字段的值Hash后放入某个桶中
(4)对于数据抽样、特定join的优化很有意义
(5)导入数据方式一般采取 INSERT … FROM … cluster by …

七、数据操作
1、向管理表装载数据
LOAD DATA LOCAL INPATH ‘XXX’ OVERWRITE INTO TABLE employees PARTITION (country=‘us’,state=‘ca’);
如果分区目录不存在，此命令会先创建分区目录，然后再将数据拷贝到该目录下。
注意：(1)LOAD DATA LOCAL 拷贝数据；LOAD DATA（没有LOCAL）转移数据（相当于剪切）
(2)指定OVERWRITE，会删除旧数据，再把新数据导入
(3)没有指定OVERWRITE，如果目标文件夹下已经存在同名的文件，会保留之前的文件，此次导入的文件会重命名。
(4)HIVE并不会验证用户装载的数据和表的模式是否匹配，但会验证文件格式是否和表结构定义的一致。如，在
创建表时定义的存储格式是SEQUENCEFILE，则加载的文件也需要是sequencefile格式。
2、通过查询语句向表中插入数据
(1)INSERT OVERWRITE TABLE employees PARTITION (country = ‘US’, state = ‘OR’)
SELECT * FROM staged_employees se WHERE se.cnty=‘US’ AND se.st=‘OR’;
(2)FROM staged_employees se
INSERT OVERWRITE TABLE employees PARTITION (country = ‘US’, state = ‘OR’) SELECT * WHERE se.cnty=‘US’ AND se.st=‘OR’
INSERT OVERWRITE TABLE employees PARTITION (country = ‘US’, state = ‘CA’) SELECT * WHERE se.cnty=‘US’ AND se.st=‘CA’;
(3)动态分区插入
INSERT OVERWRITE TABLE employees PARTITION (country, state) SELECT …, se.cnty,se.st FROM staged_employees se;
INSERT OVERWRITE TABLE employees PARTITION (country=‘US’, state) SELECT …, se.cnty,se.st FROM staged_employees se where se.cnty=‘US’;
静态分区键必须出现在动态分区键之前。动态分区功能默认情况下没有开启，开启后默认以”严格“模式执行，这种模式要求至少有一列分区字段是静态的。
3、单个查询语句中创建表并加载数据
CREATE TABLE ca_employees AS SELECT name FROM employees;
4、导出数据
INSERT OVERWRITE DIRECTORY ‘${hivevar:EXPORT_HDFS_PATH}’
ROW FORMAT SERDE ‘org.apache.hadoop.hive.contrib.serde2.MultiDelimitSerDe’
WITH SERDEPROPERTIES (‘field.delim’=’!&’,‘escape.delim’=’//’,‘serialization.null.format’=’’)
SELECT * FROM 表名;

八、排序
1、order by：全局排序，但是将所有的数据全部发送到一个Reduce中，所以在大数据量的情况下可能不能接受，最后这个操作将会产生一个文件。
2、sort by：只能保证在同一个reduce中的数据可以按指定字段排序。使用sort by你可以指定执行的reduce个数 （set mapreduce.job.reduce=）
这样可以输出更多的数据。对输出的数据再执行归并排序，即可以得到全部结果。需要注意的是，N个Reduce处理的数据范围是可以重叠的，所以
最后排序完的N个文件之间数据范围是有重叠的。
3、distribute by：控制map端的输出在reducer中是如何划分的。
distribute by rand() //将数据随机分到每个reduce上
4、cluster by：是distribute by和sort by结合，不过sort by默认是升序(ASC)，不能选择降序(DESC)。

九、Join
1、Reduce Side Join
Reduce side join是一种最简单的join方式，又叫common join。map阶段的主要任务是对不同文件中的数据打标签。
reduce阶段进行实际的连接操作。
之所以存在reduce side join，是因为在map阶段不能获取所有需要的join字段，即：同一个key对应的字段可能位于不同map中。
Reduce side join是非常低效的，因为shuffle阶段要进行大量的数据传输。
2、Map Side Join
Map side join是针对以下场景进行的优化：两个待连接表中，有一个表非常大，而另一个表非常小，以至于小表可以直接存放到内存中。
这样，我们可以将小表复制多份，让每个map task内存中存在一份（比如存放到hash table中），然后只扫描大表：对于大表中的每一条记录key/value，
在hash table中查找是否有相同的key的记录，如果有，则连接后输出即可。
3、Semi Join
(1)Semi Join，也叫半连接，是从分布式数据库中借鉴过来的方法。它的产生动机是：对于reduce side join，跨机器的数据传输量非常大，
这成了join操作的一个瓶颈，如果能够在map端过滤掉不会参加join操作的数据，则可以大大节省网络IO。
(2)实现方法很简单：选取一个小表，假设是File1，将其参与join的key抽取出来，保存到文件File3中，File3文件一般很小，可以放到内存中。在map阶段，
使用DistributedCache将File3复制到各个TaskTracker上，然后将File2中不在File3中的key对应的记录过滤掉，剩下的reduce阶段的工作与reduce side join相同。
4、Reduce Side Join + Bloom Filter
(1)在某些情况下，SemiJoin抽取出来的小表的key集合在内存中仍然存放不下，这时候可以使用BloomFiler以节省空间。
(2)BloomFilter最常见的作用是：判断某个元素是否在一个集合里面。它最重要的两个方法是：add() 和contains()。最大的特点是不会存在 false negative，
即：如果contains()返回false，则该元素一定不在集合中，但会存在一定的 false positive，即：如果contains()返回true，则该元素一定可能在集合中。
(3)因而可将小表中的key保存到BloomFilter中，在map阶段过滤大表，可能有一些不在小表中的记录没有过滤掉（但是在小表中的记录一定不会过滤掉），
这没关系，只不过增加了少量的网络IO而已

十、数据倾斜的常见场景
任务进度长时间维持在99%（或100%），查看任务监控页面，发现只有少量（1个或几个）reduce子任务未完成。
1、空值(或者某个特定值)数据倾斜
将空值(或者某个特定值)赋新的随机值
select * from t1 left outer join t2 on case when t1.k1 is null then concat(‘xxxx’,rand()) esle t1.k1 end = t2.k2;
select * from t1 left outer join t2 on case when t1.k1=0 then floor(rand()*100000) esle t1.k1 end = t2.k2;
2、不同数据类型关联产生数据倾斜
如t1的k1类型为string，t2的k2类型为bigint，两者进行关联，可能出现数据倾斜。

十一、map task 数
一个map task 处理一个input split；即多少个InputSplit就有多少个map任务。
1、Input Split的划分方式是由用户自定义的InputFormat决定的，默认情况下，有以下三个参数决定。
mapred.min.split.size ：Input Split的最小值 默认值1
mapred.max.split.szie: Input Split的最大值
dfs.block.size：HDFS 中一个block大小 默认值64MB
golsize:它是用户期望的Input Split数目=totalSize/numSplits ,其中totalSize为文件的总大小；
numSplits为用户设定的Map Task个数，numSplits=mapred.map.tasks ，默认情况下是1。
计算公式：
goalSize = totalSize /mapred.map.tasks
SplitSize = max(mapred.min.split.size,min(goalSize,blockSize))
注意：一个input split只能包含一个文件，因此SplitSize大小能平衡map task处理数据的大小。
2、合并小文件，减少map数
set mapred.max.split.size=100000000;（100M）//每个map处理的最大的文件大小，单位为B
set mapred.min.split.size.per.node=100000000;(100M) //节点中可以处理的最小的文件大小
set mapred.min.split.size.per.rack=100000000;(100M) //机架中可以处理的最小的文件大小
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; //表示执行前进行小文件合并
3、适当的增加map数
如果一个表t1只有一个文件，且较大，则考虑将一个文件合理拆成多个，这样就可以由多个map task完成，而不是只有一个。
思路：
set mapred.reduce.tasks=10; //设置reduce任务为10个
create t1_tmp as select * from t1 distribute by rand(123);
这样将t1表中的记录，随机分散到包含10个文件的t1_tmp表中，再执行下面逻辑，则会用10个map任务完成。
Select data_desc,count(1),sum(…) from t1_tmp group by data_desc；
4、控制map数量需要遵循两个原则：使大数据量利用合适的map数；使单个map任务处理合适的数据量。

十二、reduce task 数
确定reduce task数是基于以下两个参数设置：
hive.exec.reducers.bytes.per.reducer默认为1000000000(100M)
hive.exec.reducers.max（默认为999）
计算公式：
N=min(hive.exec.reducers.max，总输入数据量/hive.exec.reducers.bytes.per.reducer)

十三、LEFT SEMI JOIN
Hive不支持where子句中的子查询，SQL常用的exist in子句需要改写。
SELECT a.key,a.value from a where a.key in(SELECT b.key from b) 改写成
SELECT a.key,a.value from a left semi join b on (a.key = b.key)

十四、合并小文件
文件数目过多，会给 HDFS 带来压力，并且会影响处理效率，可以通过合并 Map 和 Reduce 的结果文件来消除这样的影响：
hive.merge.mapfiles = true --是否和并 Map 输出文件，默认为 True
hive.merge.mapredfiles = false --是否合并 Reduce 输出文件，默认为 False
hive.merge.size.per.task = 25610241024 --合并文件的大小

十五、开启本地Map任务
当以下三个参数同时成立时候，才会采用本地任务。
set hive.exec.mode.local.auto=true; //开启本地模式
set hive.exec.mode.local.auto.inputbytes.max=50000000; //设置local 的最大输入数据量,当输入数据量小于这个值的时候会采用local 的方式
set hive.exec.mode.local.auto.tasks.max=10; //设置local 的最大输入文件个数,当输入文件个数小于这个值的时候会采用local 的方式
除此外，下面参数决定本地任务可以使用的内存百分比，如果超出此内存大小，也不会执行本地任务。
hive.mapjoin.localtask.max.memory.usage //默认值： 0.90

十六、优化limit
执行limit时，很多情况下会执行全表查询，而只返回很少一部分数据，所以这种操作很浪费时间。
Set hive.limit.optimize.enable = true;
这个参数保证hive使用limit查询时进行抽样查询，不需要进行全表查询，节省很多时间。缺点是有些需要的数据可能被忽略掉（抽样）
以下两个参数配合使用：
hive.limit.row.max.size //每一行最大长度
hive.limit.optimize.limit.file //从多少个数据文件中进行抽样

十七、JVM重用
JVM重利用可以是JOB长时间保留slot，直到作业结束，这在对于有较多任务和较多小文件的任务是非常有意义的，减少执行时间。当然这个值
不能设置过大，因为有些作业会有reduce任务，如果reduce任务没有完成，则map任务占用的slot不能释放，其他的作业可能就需要等待。
Set mapred.job.reuse.jvm.num.tasks = 10; --How many tasks to run per jvm. If set to -1, there is no limit.

十八、推测式执行（Speculative Execution）
推测式执行时Hadoop的一个重要的功能，其目的是大大加快hadoop作业的整体执行时间。这个功能需要设置mapred-site.xml中的两个参数来激活这一功能：
Set mapred.map.tasks.speculative.execution = true;
Set mapred.reduce.tasks.speculative.execution = true;
Hive提供了一个参数来控制reduce端的推测式执行：
Set hive.mapred.reduce.tasks.speculative.execution = true;

十九、Group By
1、Map 端部分聚合：并不是所有的聚合操作都需要在 Reduce 端完成，很多聚合操作都可以先在 Map 端进行部分聚合，最后在 Reduce 端得出最终结果。
参数包括：
hive.map.aggr = true //是否在 Map 端进行聚合，默认为 True
hive.groupby.mapaggr.checkinterval = 100000 //在 Map 端进行聚合操作的条目数目
2、有数据倾斜的时候进行负载均衡
hive.groupby.skewindata = true
当选项设定为 true，生成的查询计划会有两个MR Job。第一个MR Job中，Map的输出结果集合会随机分布到Reduce中，每个Reduce做部分聚合操作，
并输出结果，这样处理的结果是相同的Group By Key有可能被分发到不同的Reduce中，从而达到负载均衡的目的；第二个 MR Job 再根据第一个
MR JOB的数据结果按照Group By Key分布到Reduce（这个过程可以保证相同的Group By Key被分布到同一个Reduce中），最后完成最终的聚合操作。
注意：当启用时，能够解决数据倾斜的问题，但如果要在查询语句中对多个字段进行去重统计时会报错。
hive> set hive.groupby.skewindata=true;
hive> select count(distinct id),count(distinct x) from test;
FAILED: SemanticException [Error 10022]: DISTINCT on different columns not supported with skew in data
下面这种方式是可以正常查询
hive>select count(distinct id, x) from test;

二十、慎用count(distinct)
1、SELECT COUNT( DISTINCT id ) FROM TABLE_NAME WHERE …;
由于引入了DISTINCT，因此在Map阶段无法利用combine对输出结果消重，必须将id作为Key输出，在Reduce阶段只有一个reduce再对来自于不同Map Task，
相同Key的结果进行消重，计入最终统计值。
因为Hive在处理COUNT这种“全聚合(full aggregates)”计算时，它会忽略用户指定的Reduce Task数(mapred.reduce.tasks)，而强制使用1个reduce。
2、改进写法：SELECT COUNT() FROM (SELECT DISTINCT id FROM TABLE_NAME WHERE … ) t;
改进后的写法会有两个MR作业，第一个MR作业选出全部的非重复id，可以通过增大Reduce的并发数，并发处理Map输出。第二个MR作业再对这些已消重的id进行计数。
由于id已经消重，因此COUNT()操作在Map阶段不需要输出原id数据，只输出一个合并后的计数即可。
实际运行时，Hive还对这两阶段的作业做了额外的优化。它将第二个MapReduce作业Map中的Count过程移到了第一个作业的Reduce阶段。
这样在第一阶Reduce就可以输出计数值，而不是消重的全部id。这一优化大幅地减少了第一个作业的Reduce输出IO以及第二个作业Map的输入数据量。

二十一、使用ORC File
ORC File，它的全名是Optimized Row Columnar (ORC) file，其实就是对RCFile做了一些优化。据官方文档介绍，这种文件格式可以提供一种高效的方法来存储Hive数据。
它的设计目标是来克服Hive其他格式的缺陷。运用ORC File可以提高Hive的读、写以及处理数据的性能。输出单个文件，这样可以减少NameNode的负载；
(2)、支持各种复杂的数据类型，比如： datetime, decimal, 以及一些复杂类型(struct, list, map, and union)；
(3)、在文件中存储了一些轻量级的索引数据；
(4)、基于数据类型的块模式压缩：a、integer类型的列用行程长度编码(run-length encoding);b、String类型的列用字典编码(dictionary encoding)；
(5)、用多个互相独立的RecordReaders并行读相同的文件；
(6)、无需扫描markers就可以分割文件；
(7)、绑定读写所需要的内存；
(8)、metadata的存储是用 Protocol Buffers的，所以它支持添加和删除一些列。

二十二、使用并发
1、hive.exec.parallel=true; //允许单个Hive语句生成的多个MR任务根据依赖关系并行执行。
task1和task2之间没有依赖关系，两个任务又都会依赖task3，正常情况下hive会按照task1->task2->task3的顺序执行；
在设置hive.exec.parallel=true之后，hive会同时执行task1和task2，当两个任务都完成之后，再开始task3。
2、hive.exec.parallel.thread.number=32; //允许最大并发的任务数量，仅在hive.exec.parallel=true的时候生效。
优点：将MR任务进行并发以提升运行速度。
缺点：1、使用条件较为严格，仅限于较复杂的SQL，即由一个SQL可以生成多个MR任务，
并且任务的依赖关系之间存在并行关系的情况下才可以生效。所以不适合简单语句的场景。
2、多并发场景下会占用Yarn环境资源，影响其他任务执行。

二十三、Uber模式
1、原理：正常情况下，MR任务的执行流程为：任务提交提交之后，启动AppMaster进程来进行单个任务的调度管理，之后为每个Map和Reduce任务启动YarnChild进程用来执行任务，
每个Child进程只执行一个Map或者Reduce任务，执行完毕就退出，所以正常情况下，一个MR任务可能会启动多个进程。
Uber模式下，用户提交的任务，都会在AppMaster进程中线性执行，整个任务运行期间，只会启动一个进程。这样就通过JVM重用来减少了进程启动和销毁的时间。
2、使用方法
set mapreduce.job.ubertask.enable=true;
set mapreduce.job.ubertask.maxmaps=9;
set mapreduce.job.ubertask.maxreduces=1;

二十四、数据类型使用
1、Structs： structs内部的数据可以通过DOT（.）来存取，例如，表中一列c的类型为STRUCT{a INT; b INT}，我们可以通过c.a来访问域a
定义方式：info struct<name:STRING, age:INT>
2、Maps（K-V对）：访问指定域可以通过[“指定域名称”]进行，例如，一个Map M包含了一个group-》gid的kv对，gid的值可以通过M[‘group’]来获取
定义方式：id array
3、Arrays：array中的数据为相同类型，例如，假如array A中元素[‘a’,‘b’,‘c’]，则A[1]的值为’b’
perf map<string, int>
4、指定分隔符
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ‘\t’
COLLECTION ITEMS TERMINATED BY ‘,’
MAP KEYS TERMINATED BY ‘:’;
5、split(string str, string pat) 将str按pat分隔，划分成数组
6、COALESCE(T v1, T v2, …) 返回参数中的第一个非空值；如果所有值都为 NULL，那么返回NULL。
7、collect_list(col) 列出该字段所有的值，不去重 select collect_list(id) from table;
8、collect_set(col) 列出该字段所有的值，去重

二十五、UDF
1、简单UDF： extends UDF ， 仅重写evaluate方法
(1)必须继承org.apache.hadoop.hive.ql.exec.UDF
(2)必须重写evaluate函数，用户在该函数内自定义数据的处理流程。evaluate函数支持重入，入参和返回值类型都可以由用户定义。
2、通用UDF：需要重写initialize、getDisplayString、evaluate方法；简单UDF使用java的反射机制，会导致性能的损失。
(1)通用UDF必须继承org.apache.hadoop.hive.ql.udf.generic.GenericUDF
(2)通用UDF必须重写三个方法，initialize方法，evaluate方法和getDisplayString方法，且这些方法均不支持重载。
(3)initialize方法：用于函数初始化操作，并定义函数的返回值类型。
(4)evaluate方法：具体处理用户UDF逻辑。
(5)getDisplayString方法：返回的是函数的描述信息，与具体的处理逻辑无关。
3、使用方法：
(1)将自定义的函数类导出到jar包
(2)将jar包放在主HiveServer所在节点上
(3)进入hive控制台，执行add jar操作，使HiveServer加载jar包。add jar hdfs://hacluster/xxx/udf.jar;
(4)执行create function命令创建自定义函数。 CREATE TEMPORARY FUNCTION 函数名 AS ‘xxx.xxx’;
(5)创建成功后即可在HQL语句中直接使用该函数。

二十六、UDAF
1、UDAF的编写需要实现两层类，函数类和内部类，函数类需要继承一个GenericUDAFResolver类，内部类是具体的Evaluator，定义了具体的处理逻辑，
该类需要实现一个GenericUDAFEvaluator接口。
2、函数类Resolver必须实现getEvaluator方法，getEvaluator方法用于获取具体的Evaluator，一般是内部类对象
3、内部类Evaluator需要实现 init、iterate、terminatePartial、merge、terminate这几个函数。
(1)init方法：确定各个阶段输入输出参数的数据格式ObjectInspectors
(2)iterate方法：在map阶段调用，迭代处理传入的参数，一个map处理每一行时都会调用一次该方法。
(3)terminatePartial方法：map与combiner结束返回结果，得到部分数据聚集结果
(4)merge方法：接收terminatePartial方法的返回结果，在combiner阶段合并map阶段返回的结果，或者在reducer阶段合并mapper阶段或combiner阶段返回的结果。
(5)terminate方法：入参为merge方法处理后的buffer，返回最终的聚集函数结果，一般在reduce阶段调用，如果没有reduce，会在map端调用该函数返回最终结果。
4、Mode代表了UDAF在mapreduce的各个阶段。在init方法的输入参数中有Mode对象。
(1)PARTIAL1: 这个是mapreduce的map阶段，从原始数据到部分数据聚合，将会调用iterate()和terminatePartial()
(2)PARTIAL2: 这个是mapreduce的map端的Combiner阶段，负责在map端合并map的数据:从部分数据聚合到部分数据聚合，将会调用merge()和terminatePartial()
(3)FINAL: mapreduce的reduce阶段，从部分数据的聚合到完全聚合，将会调用merge()和terminate()
(4) COMPLETE: 如果出现了这个阶段，表示mapreduce只有map，没有reduce，所以map端就直接出结果了，从原始数据直接到完全聚合，将会调用 iterate()和terminate()

二十七、UDTF
用于处理一列输入，多列输出或者一行输入多行输出的场景。
1、UDTF必须继承org.apache.hadoop.hive.ql.udf.generic.GenericUDTF。
2、需要重写initialize, process, close三个方法。
(1)initialize方法：此方法返回UDTF的返回行的信息（返回个数，类型），也就是返回一个处理返回值类型的Inspector对象。
(2)process方法：该方法定义了真正的数据处理过程，在process中，每一次forward()调用都会产生一行输出；如果产生多列可以将多个列的值放在一个数组中，
然后将该数组传入到forward()函数。如果需要产生多行，可以将一行分解，然后多次调用forward()函数。
(3)close()方法：对需要清理的方法进行清理。

二十八、经验
1、检查数据倾斜
select 字段，count(1) as cn from 表名 group by 字段 order by cn desc limit 10;
2、hive去掉重复记录，即对于重复记录，只取其中一行
select * from
(
select cust_id,act_nbr,row_number() over (partition by cust_id order by act_nbr desc) rn from t_act_inf
) aa where rn=1;
3、set hive.exec.compress.output=false;
这是设置 insert overwrite directory 到指定路径时，不进行压缩；
因为当文件比较大时，hive有可能会进行压缩，则不能通过getmerge来获得文件，
因为会得到压缩的文件，从而看到的文件会有乱码；
但可以text来查看文件，并重定向到指定文件，因为text会解压缩；
4、hdfs dfs -getmerge [nl]
！