Hive使用简单总结

一.Hive介绍

1.1.Hive概述

Hive是基于Hadoop的一个数据仓库工具。可以将结构化的数据文件映射为一张表，并提供完整的sql查询功能，可以将sql语句转换为MapReduce任务进行运行。其优点是学习成本低，可以通过类SQL语句快速实现MapReduce统计，不必开发专门的MapReduce应用，十分适合数据仓库的统计分析。
Hive是建立在 Hadoop 上的数据仓库基础构架。它提供了一系列的工具，可以用来进行数据提取、转化、加载（ETL Extract-Transform-Load ），这是一种可以存储、查询和分析存储在 Hadoop 中的大规模数据的机制。Hive 定义了简单的类 SQL 查询语言，称为 HiveQL，它允许熟悉 SQL 的用户查询数据。
1.1.1 数据仓库的特征
1.数据仓库是多个异构数据源所集成的。
2.数据仓库存储的一般是历史数据。大多数的应用场景是读数据（分析数据），所以数据仓库是弱事务的。
3.数据库是为捕获数据而设计，数据仓库是为分析数据而设计。
4.数据仓库是时变的，数据存储从历史的角度提供信息。即数据仓库中的关键结构都隐式或显示地包含时间元素。
5.数据仓库是弱事务的，因为数据仓库存的是历史数据，一般都读（分析）数据场景。
1.1.2 OLAP和OLTP
数据库属于OLTP系统。（Online Transaction Processing）联机事务处理系统。涵盖了企业大部分的日常操作，如购物、库存、制造、银行、工资、注册、记账等。比如Mysql，oracle等关系型数据库。
数据仓库属于OLAP系统。（Online Analytical Processing）联机分析处理系统。Hive，Hbase等。
OLTP是面向用户的、用于程序员的事务处理以及客户的查询处理。
OLAP是面向市场的，用于知识工人（经理、主管和数据分析人员）的数据分析。
OLAP通常会集成多个异构数据源的数据，数量巨大。
OLTP系统的访问由于要保证原子性，所以有事务机制和恢复机制。
OLAP系统一般存储的是历史数据，所以大部分都是只读操作，不需要事务。

1.2.适用场景

Hive 构建在基于静态（离线）批处理的Hadoop 之上，Hadoop 通常都有较高的延迟并且在作业提交和调度的时候需要大量的开销。因此，Hive 并不能够在大规模数据集上实现低延迟快速的查询，例如，Hive 在几百MB 的数据集上执行查询一般有分钟级的时间延迟。因此，Hive 并不适合那些需要低延迟的应用，例如，联机事务处理(OLTP)。Hive 查询操作过程严格遵守Hadoop MapReduce 的作业执行模型，Hive 将用户的HiveQL 语句通过解释器转换为MapReduce 作业提交到Hadoop 集群上，Hadoop 监控作业执行过程，然后返回作业执行结果给用户。Hive 并非为联机事务处理而设计，Hive 并不提供实时的查询和基于行级的数据更新操作。Hive 的最佳使用场合是大数据集的离线批处理作业，例如，网络日志分析。

二. HiveSQL

按数据操作分类，来分别说明脚本的使用。

2.1. DDL

2.1.1 建表说明
元数据：描述数据的数据
表分类：主要分内表和外表
内表：元数据和数据本身均被hive管理。删除表则全部删除。
外表：元数据被hive管理，数据本身存储在hdfs,不受hive管理。删除表则只删除元数据，数据本身不变。
2.1.2 建表模板
CREATE [external] TABLE [IF NOT EXISTS] table_name
[(col_name data_type [comment col_comment], …)]
[comment table_comment]
[partitioned by (col_name data_type [comment col_comment], …)]
[clustered by (col_name, col_name, …)
[sorted by (col_name [ASC|DESC], …)] INTO num_buckets BUCKETS]
[row format row_format]
[stored as file_format]
[location hdfs_path]
关键词解释
• external: 创建内部表还是外部表，此为内外表的唯一区分关键字。
• comment col_comment: 给字段添加注释
• comment table_comment: 给表本身添加注释
• partitioned by: 按哪些字段分区，可以是一个，也可以是多个
• clustered by col_name… into num_buckets BUCKETS:按哪几个字段做hash后分桶存储
• row format:用于设定行、列、集合的分隔符等设置
• stored as : 用于指定存储的文件类型，如text,rcfile等
• location : 设定该表存储的hdfs目录，如果不手动设定，则采用hive默认的存储路径
2.1.3 示例
示例：创建学生表student,包括id,name,classid,classname及分区和注释信息。
CREATE TABLE student(
id string comment ‘学号’,
username string comment ‘姓名’,
classid int comment ‘班级id’,
classname string comment ‘班级名称’
)
comment ‘学生信息主表’
partitioned by (come_date string comment ‘按入学年份分区’)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ‘\001’
LINES TERMINATED BY ‘\n’
STORED AS textfile;

2.2. DML

2.2.1 加载数据脚本
LOAD DATA [LOCAL] INPATH ‘filepath’ [OVERWRITE] INTO TABLE tablename [PARTITION (partcol1=val1, partcol2=val2 …)]
2.2.2 加载本地数据文件
创建一个文本文件存储的表，并以"\t"作为分隔符，方便构造和上传数据
CREATE TABLE student(
id string comment ‘学号’,
username string comment ‘姓名’,
classid int comment ‘班级id’,
classname string comment ‘班级名称’
)
comment ‘学生信息主表’
partitioned by (come_date string comment ‘按入学年份分区’)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ‘\t’
LINES TERMINATED BY ‘\n’
STORED AS textfile;
2.2.3 将查询结果插入到数据表中
• 脚本模板
INSERT OVERWRITE TABLE tablename1
[PARTITION (partcol1=val1, partcol2=val2 …)]
select_statement1 FROM from_statement
• 样例
insert overwrite table student partition(come_date=‘20170905’)
select
id,username,classid,classname
from student
where come_date=‘20190923’;
2.2.4 多插入模式（一次查询多次插入）
• 模板
FROM from_statement
INSERT OVERWRITE TABLE tablename1 [PARTITION (partcol=val)] select_statement1
[INSERT OVERWRITE TABLE tablename2 [PARTITION …] select_statement2
…
• 样例
from student
insert overwrite table student partition(come_date=‘20170921’) select id,username,classid,classname where come_date=‘20190921’
insert overwrite table student partition(come_date=‘20170922’) select id,username,classid,classname where come_date=‘20190922’
insert overwrite table student partition(come_date=‘20170923’) select id,username,classid,classname where come_date=‘20190923’
2.2.5 动态分区模式（让分区成为变量）
• 脚本模板
INSERT OVERWRITE TABLE tablename
PARTITION (partcol1[=var1], partcol2[=var2] …) select_statement FROM from_statement
• 设置非严格模式
set hive.exec.dynamic.partition.mode=nonstric;
• 样例
u 有新表student_outer，即新来的外部学生表，有若干学生数据，与student表结构完全相同。
u 将student_outer表的多个分区数据，一次性插入到student表中。
• 数据准备
u 创建新表student_outer与之前的student结构完全一致
CREATE TABLE student_outer(
id string comment ‘学号’,
username string comment ‘姓名’,
classid int comment ‘班级id’,
classname string comment ‘班级名称’
)
comment ‘学生信息主表’
partitioned by (come_date string comment ‘按入学年份分区’)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ‘\t’ LINES TERMINATED BY ‘\n’
STORED AS textfile;
• 装载数据
LOAD DATA LOCAL INPATH ‘./student.txt’ OVERWRITE INTO TABLE student_outer PARTITION (come_date=20190921);
LOAD DATA LOCAL INPATH ‘./student.txt’ OVERWRITE INTO TABLE student_outer PARTITION (come_date=20191122);
LOAD DATA LOCAL INPATH ‘./student.txt’ OVERWRITE INTO TABLE student_outer PARTITION (come_date=20191123);
• 将student_outer表所有数据插入到student表
//先清空之前的student表，方便查看效果
truncate table student;
insert overwrite table student partition(come_date)
select id,username,classid,classname,come_date from student_outer
2.2.6 将查询结果写入hdfs目录
• 脚本模版
INSERT OVERWRITE [LOCAL] DIRECTORY directory1 SELECT … FROM …
• 样例-默认
数据写入文件系统时进行文本序列化，且每列用^A来区分，\n换行
insert overwrite directory “/tmp/output2/”
select * from student where come_date=‘20190923’;
• 样例-指定输分隔符
insert overwrite directory “/tmp/output2/”
row format delimited
fields terminated by ‘\t’
select * from student where come_date=‘20190923’;
2.2.7 关于外表的说明和使用
• 内外表的唯一区分标识，即为external关键字。创建表时候有，则为外表，没有则默认均为内表。
• 内表和外表的hdfs目录，均可以自由指定location，如不指定，则数据存储在hive的默认hdfs目录中，且后续均可以自由改变，但改变的场景不多，一般为了省事都不会轻易改变存储目录。
• 示例
创建外表
CREATE external TABLE student_external(
id string comment ‘stdno’,
username string comment ‘name’,
classid int comment ‘class id’,
classname string comment ‘class name’
)
comment ‘学生信息主表’
partitioned by
(come_date string comment ‘come to school date’)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ‘\t’
LINES TERMINATED BY ‘\n’
STORED AS textfile;
从本地装载数据
LOAD DATA LOCAL INPATH ‘./student.txt’
OVERWRITE INTO TABLE student_external
PARTITION (come_date=20171120);
删除数据（删除数据后，hdfs文件依然存在）

2.3. DQL

2.3.1 脚本模板
SELECT [DISTINCT] select_expr, select_expr, …
FROM table_reference
[WHERE where_condition]
[GROUP BY col_list [HAVING condition]]
[ CLUSTER BY col_list
| [DISTRIBUTE BY col_list] [SORT BY| ORDER BY col_list]
]
[LIMIT number]
DISTRIBUTE BY col_list
以指定字段作为key作hash partition,保证相同的key会到同一个reduce去处理。
Sort By col_list
以指定字段作为单个reduce排序的key，保证单个reduce内的key有序排列输出。
Order By col_list
只会生成一个reduce任务，对全部排序
CLUSTER BY col_list
以指定字段作为key做hash partition，保证相同key会到同一个reduce去处理。该命令相当于distributed by col_list和sort by col_list的联合使用。
2.3.2 典型样例
查询所有记录
select * from student
加入where查询条件
select * from student where id=‘001’;
加入limit限制
select * from student where id=‘001’ limit 3;
升降序：desc,asc
2.3.3 join查询
将多个表通过字段关联在一起，形成查询结果
创建student_location表，存储学生的地理位置信息
CREATE TABLE student_location(
id string comment ‘stdno’,
province string comment ‘province name’,
city string comment ‘city name’,
region string comment ‘region name’
)
comment ‘student location info’
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ‘\t’
LINES TERMINATED BY ‘\n’
STORED AS textfile;
构造student_location表数据
001 河北省石家庄市高新区
002 河北省石家庄市桥西区
003 河南省郑州市高新区
004 湖南省长沙市高新区
005 北京市北京市朝阳区
006 北京市北京市海淀区
将数据加载进student_location表中
LOAD DATA LOCAL INPATH ‘./student_location.txt’
OVERWRITE INTO TABLE student_location;
join查询示例
• inner join
将左表和右表满足联接条件的数据，全部查询出来
select * from student
inner join student_location
on student.id=student_location.id;
• left outer join
以左表为主，将左表数据全部保留，没有关联上数据字段置成NULL
select * from student left
outer join student_location
on student.id=student_location.id;
• right outer join
以右表为主，将右表数据全部保留，没有关联上数据字段置成NULL
select * from student
right outer join student_location
on student.id=student_location.id;
• full outer join
没有关联上数据字段全部置成NULL
select * from student
full outer join student_location
on student.id=student_location.id;
2.3.4 union
• union all
将所有表数据，完全叠加在一起，不去重。
要求：所有表的字段和类型完全一致。
以student和student_outer表为例
select * from student
union all
select * from student_outer;
• union
将所有表数据，完全叠加在一起，总体去重。
要求：所有表的字段和类型完全一致。
l 以student和student_outer表为例
select * from student
union
select * from student_outer;

2.4.Hive 数据分析函数

分析函数、窗口函数、增强 Group 三类，及用于辅助表达的 over 从句。
2.4.1 产生背景
常规 SQL 语句中，明细数据和聚合后的数据不能同时出现在一张表中，而此类需求又常见。如：员工既要查询当前收入多少又要显示本年度收入多少; 员工既要查询当前收入多少，又要显示历史总收入多少; 购物者既要查询当前剩余多少，又要显示历史充值多少等等。该类函数即为解决两者可以同时出现的问题。
2.4.2 函数分类
分析函数（不支持与 window 子句联用，即 ROWS BETWEEN）
• NTILE：序列分析函数，用于数据分片排序，对数据分片并返回当前分片值。(即对数据集分块，如第 1 块，第 2 块等)，不均匀的时候，依次增加前边分片序列的数量。
• ROW_NUMBER：序列分析函数，用于排序，按照顺序，不关心是否有相等情况，从 1 开始逐条给数据一个加 1 后的序列号。如 1，2，3，4…  RANK：序列分析函数，用于排序，按照顺序，关心相等情况，如遇到相等情况，名次会留下空位。如 1,2,2,4,4,6…
• DENSE_RANK：序列分析函数，用于排序，关心相等情况，如遇到相等情况，名次不会留下空位。如 1,2,2,3,3,4…
• CUME_DIST：累计计算函数，计算公式为“小于等于当前值的行数/分组内总行数",用于计算数据分布等场景。
• PERCENT_RANK：累计计算函数，计算公式为” 分组内当前行的 RANK 值- 1/分组内总行数-1",用于计算处理进度等场景。
窗口函数
• LAG:函数 LAG(col,n,DEFAULT)用于统计窗口内往上第 n 行值。第一个参数为列名，第二个参数为往上第 n 行（可选，默认为 1），第三个参数为默认值（当往上第 n 行为 NULL 时候，取默认值，如不指定，则为 NULL）
• LEAD:与 LAG 作用相反，函数形式如 LEAD(col,n,DEFAULT) 用于统计窗口内往下第 n 行值。第一个参数为列名，第二个参数为往下第 n 行 (可选，默认为 1)，第三个参数为默认值(当往下第 n 行为 NULL 时候，取默认值，如不指定，则为 NULL)
• FIRST_VALUE: 取分组内排序后，截止到当前行，第一个值
• LAST_VALUE:与 FIRST_VALUE 相反，取分组内排序后，截止到当前行，最后一个值
增强 GroupBy
• GROUPING SETS：根据不同的维度组合进行聚合，等价于将不同维度的 GROUP BY 结果集进行 UNION ALL
• GROUPING__ID：表示结果属于哪一个分组集合，属于虚字段  CUBE：根据 GROUP BY 的维度的所有组合进行聚合。
• ROLLUP：为 CUBE 的子集，以最左侧的维度为主，从该维度进行层级聚合。
over()从句：指定分析窗口函数的细化落围规则
• 标准的聚合函数 COUNT、SUM、MIN、MAX、AVG 联用，如 sum(…) over(…)
• 与区配的分析窗口联用，如 ROW_NUMBER() over(…)
• 使用 PARTITION BY 语句，使用一个或者多个原始数据类型的列
• 使用 PARTITION BY 与 ORDER BY 语句，使用一个或者多个数据类型的分区或者排序列
• 使用窗口规范，窗口规范支持以下格式：
ROWS BETWEEN:即为 window 子句或称窗口子句，属于物理截取，即物理窗口，从行数上控制截取数据的大小多少。
RANGE BETWEEN: 即为 window 子句或称窗口子句，属于逻辑截取，即逻辑窗口，从列值上控制窗口的大小多少。
PRECEDING：window 子句之往前  FOLLOWING：window 子句之往后  CURRENT ROW：window 子句之当前行
UNBOUNDED：window 子句之起点，UNBOUNDED PRECEDING 表示从前面的起点，UNBOUNDED FOLLOWING：表示到后面的终点。
• 注意：order by 子句后边如果没有跟着多大窗口，则默认为 range between unbounded preceding and current row

2.5.Hive解决数据倾斜问题

2.5.1 概述
什么是数据倾斜以及数据倾斜是怎么产生的？
简单来说数据倾斜就是数据的key 的分化严重不均，造成一部分数据很多，一部分数据很少的局面。
举个 word count 的入门例子，它的map 阶段就是形成（“aaa”,1）的形式，然后在reduce 阶段进行 value 相加，得出 “aaa” 出现的次数。若进行 word count 的文本有100G，其中 80G 全部是 “aaa” 剩下 20G 是其余单词，那就会形成 80G 的数据量交给一个 reduce 进行相加，其余 20G 根据 key 不同分散到不同 reduce 进行相加的情况。如此就造成了数据倾斜，临床反应就是 reduce 跑到 99%然后一直在原地等着那80G 的reduce 跑完。
如此一来 80G 的 aaa 将发往同一个 reducer ，由此就可以知道 reduce 最后 1% 的工作在等什么了。
为什么说数据倾斜与业务逻辑和数据量有关？
从另外角度看数据倾斜，其本质还是在单台节点在执行那一部分数据reduce任务的时候，由于数据量大，跑不动，造成任务卡住。若是这台节点机器内存够大，CPU、网络等资源充足，跑 80G 左右的数据量和跑10M 数据量所耗时间不是很大差距，那么也就不存在问题，倾斜就倾斜吧，反正机器跑的动。所以机器配置和数据量存在一个合理的比例，一旦数据量远超机器的极限，那么不管每个key的数据如何分布，总会有一个key的数据量超出机器的能力，造成 reduce 缓慢甚至卡顿。
业务逻辑造成的数据倾斜会多很多，日常使用过程中，容易造成数据倾斜的原因可以归纳为几点：
1）group by
2）distinct count(distinct xx)
3）join
如何处理group by的数据倾斜问题
1、调优参数
set hive.groupby.skewindata=true;
hive.groupby.skewindata=true：数据倾斜时负载均衡，当选项设定为true，生成的查询计划会有两个MRJob。第一个MRJob 中，Map的输出结果集合会随机分布到Reduce中，每个Reduce做部分聚合操作，并输出结果，这样处理的结果是相同的GroupBy Key有可能被分发到不同的Reduce中，从而达到负载均衡的目的；第二个MRJob再根据预处理的数据结果按照GroupBy Key分布到Reduce中（这个过程可以保证相同的GroupBy Key被分布到同一个Reduce中），最后完成最终的聚合操作。
由上面可以看出起到至关重要的作用的其实是第二个参数的设置，它使计算变成了两个mapreduce，先在第一个中在 shuffle 过程 partition 时随机给 key 打标记，使每个key 随机均匀分布到各个 reduce 上计算，但是这样只能完成部分计算，因为相同key没有分配到相同reduce上，所以需要第二次的mapreduce,这次就回归正常 shuffle,但是数据分布不均匀的问题在第一次mapreduce已经有了很大的改善，因此基本解决数据倾斜。

2.6.Hive优化

1）map side join
mapJoin的主要意思就是，当链接的两个表是一个比较小的表和一个特别大的表的时候，我们把比较小的table直接放到内存中去，然后再对比较大的表格进行map操作。join就发生在map操作的时候，每当扫描一个大的table中的数据，就要去去查看小表的数据，哪条与之相符，继而进行连接。这里的join并不会涉及reduce操作。map端join的优势就是在于没有shuffle，在实际的应用中，我们这样设置：

set hive.auto.convert.join=true;
此外，hive有一个参数：hive.mapjoin.smalltable.filesize，默认值是25mb（其中一个表大小小于25mb时，自动启用mapjoin）
要求：在hive做join时，要求小表在前(左）

2）join语句优化
优化前
select m.cid,u.id form order m join customer u on m.cid=u.id where m.dt=’20160801’;
优化后
select m.cid,u.id from (select cid from order where dt=’20160801’)m
join customer u on m.cid = u.id
注意：Hive在做join时，小表写在前（左边）。

3）group by 优化
hive.groupby.skewindata=true
如果group by过程出现倾斜，应该设置为true

4）count distinct 优化
优化前
select count(distinct id )from tablename
优化后
select count(*) from (select distinct id from tablename)tmp;
此外，再设定一下reduce的任务数量。
注意：count这种全局计数的操作，Hive只会用一个Reduce来实现

日常统计场景中，我们经常会对一段时期内的字段进行消重并统计数量，SQL语句类似于
SELECT COUNT( DISTINCT id ) FROM TABLE_NAME WHERE …;
这条语句是从一个表的符合WHERE条件的记录中统计不重复的id的总数。
该语句转化为MapReduce作业后执行示意图如下，图中还列出了我们实验作业中Reduce阶段的数据规模：
由于引入了DISTINCT，因此在Map阶段无法利用combine对输出结果消重，必须将id作为Key输出，在Reduce阶段再对来自于不同Map Task、相同Key的结果进行消重，计入最终统计值。
我们看到作业运行时的Reduce Task个数为1，对于统计大数据量时，这会导致最终Map的全部输出由单个的ReduceTask处理。这唯一的Reduce Task需要Shuffle大量的数据，并且进行排序聚合等处理，这使得它成为整个作业的IO和运算瓶颈。
经过上述分析后，我们尝试显式地增大Reduce Task个数来提高Reduce阶段的并发，使每一个Reduce Task的数据处理量控制在2G左右。具体设置如下：
set mapred.reduce.tasks=100
调整后我们发现这一参数并没有影响实际Reduce Task个数，Hive运行时输出“Number of reduce tasks determined at compile time: 1”。
原因是Hive在处理COUNT这种“全聚合(full aggregates)”计算时，它会忽略用户指定的Reduce Task数，而强制使用1。
所以我们只能采用变通的方法来绕过这一限制。我们利用Hive对嵌套语句的支持，将原来一个MapReduce作业转换为两个作业，在第一阶段选出全部的非重复id，在第二阶段再对这些已消重的id进行计数。这样在第一阶段我们可以通过增大Reduce的并发数，并发处理Map输出。在第二阶段，由于id已经消重，因此COUNT()操作在Map阶段不需要输出原id数据，只输出一个合并后的计数即可。这样即使第二阶段Hive强制指定一个Reduce Task，极少量的Map输出数据也不会使单一的Reduce Task成为瓶颈。改进后的SQL语句如下：
SELECT COUNT() FROM (SELECT DISTINCT id FROM TABLE_NAME WHERE … ) t;
这一优化使得在同样的运行环境下，优化后的语句执行只需要原语句20%左右的时间。优化后的MapReduce作业流如下：

5）调整切片数（map任务数）
Hive底层自动对小文件做了优化，用了CombineTextInputFormat，将做个小文件切片合成一个切片。
合成完之后的切片大小，如果>mapred.max.split.size 的大小，就会生成一个新的切片。
mapred.max.split.size 默认是128MB
set mapred.max.split.size=134217728（128MB)
对于切片数（MapTask）数量的调整，要根据实际业务来定，比如一个100MB的文件
假设有1千万条数据，此时可以调成10个MapTask，则每个MapTask处理1百万条数据。

6）JVM重利用
set mapred.job.reuse.jvm.num.tasks=20(默认是1个）
JVM重用是hadoop调优参数的内容，对hive的性能具有非常大的影响，特别是对于很难避免小文件的场景或者task特别多的场景，这类场景大多数执行时间都很短。这时JVM的启动过程可能会造成相当大的开销，尤其是执行的job包含有成千上万个task任务的情况。
JVM重用可以使得一个JVM进程在同一个JOB中重新使用N次后才会销毁。

7）启用严格模式
在hive里面可以通过严格模式防止用户执行那些可能产生意想不到的不好的效果的查询,从而保护hive的集群。
用户可以通过 set hive.mapred.mode=strict 来设置严格模式，改成unstrict则为非严格模式。
在严格模式下，用户在运行如下query的时候会报错：
①分区表的查询没有使用分区字段来限制
②使用了order by 但没有使用limit语句。（如果不使用limit，会对查询结果进行全局排序，消耗时间长）
③产生了笛卡尔积
当用户写代码将表的别名写错的时候会引起笛卡尔积，例如
SELECT *
FROM origindb.promotion__campaign c
JOIN origindb.promotion__campaignex ce
ON c.id = c.id
limit 1000

8）关闭推测执行机制
因为在测试环境下我们都把应用程序跑通了，如果还加上推测执行，如果有一个数据分片本来就会发生数据倾斜，执行执行时间就是比其他的时间长，那么hive就会把这个执行时间长的job当作运行失败，继而又产生一个相同的job去运行，后果可想而知。可通过如下设置关闭推测执行：
set mapreduce.map.speculative=false
set mapreduce.reduce.speculative=false
set hive.mapred.reduce.tasks.speculative.execution=false