order表数据：
1 20180710 P001 20
2 20180710 P002 14
3 20180710 P001 35
4 20180710 P002 40
5 20180710 P001 10
6 20180710 P003 20
7 20180710 P004 12
product表数据：
P001 xiaomi 2999
P002 huawei 3999
P005 chuizi 4000
建表：
1）create external table order_t (id string,time string,pid string,amount int) row format delimited fields terminated by ’ ’ location ‘/order’;
2）create external table product_t (pid string,name string,price int) row format delimited fields
terminated by ’ ’ location ‘/product’;
查询：
select * from product_t join order_t on product_t.pid=order_t.pid;
注意：把小表写在前面，hive底层自动做map side join.
inner join
select * from product_t inner join order_t on product_t.pid=order_t.pid;
left join
select * from product_t left join order_t on product_t.pid=order_t.pid;
right join
select * from product_t right join order_t on product_t.pid=order_t.pid;
Full outer join
select * from product_t full outer join order_t on product_t.pid=order_t.pid;
left semi join
select * from product_t left semi join order_t on product_t.pid=order_t.pid;
这种join解决的是exist in（是否存在） 的问题
a表里哪些数据在b表中出现过
概述
什么是数据倾斜以及数据倾斜是怎么产生的？
简单来说数据倾斜就是数据的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已经有了很大的改善，因此基本解决数据倾斜。
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时，要求小表在前(左）
优化前
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时，小表写在前（左边）。
hive.groupby.skewindata=true
如果group by过程出现倾斜，应该设置为true
优化前
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作业流如下：
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百万条数据。
set mapred.job.reuse.jvm.num.tasks=20(默认是1个）
JVM重用是hadoop调优参数的内容，对hive的性能具有非常大的影响，特别是对于很
难避免小文件的场景或者task特别多的场景，这类场景大多数执行时间都很短。这时JVM的启动过程可能会造成相当大的开销，尤其是执行的job包含有成千上万个task任 务的情况。
JVM重用可以使得一个JVM进程在同一个JOB中重新使用N次后才会销毁。
在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
因为在测试环境下我们都把应用程序跑通了，如果还加上推测执行，如果有一个数据分 片本来就会发生数据倾斜，执行执行时间就是比其他的时间长，那么hive就会把这个执 行时间长的job当作运行失败，继而又产生一个相同的job去运行，后果可想而知。可通 过如下设置关闭推测执行：
set mapreduce.map.speculative=false
set mapreduce.reduce.speculative=false
set hive.mapred.reduce.tasks.speculative.execution=false
1.创建带桶的 table ：
create table teacher(name string) clustered by (name) into 3 buckets row format delimited fields terminated by ’ ';
2.开启分桶机制：
set hive.enforce.bucketing=true; 3.往表中插入数据：
insert overwrite table teacher select * from tmp;//需要提前准备好temp，从temp查询数据写入到teacher
注：teacher是一个分桶表，对于分桶表，不允许以外部文件方式导入数据，只能从另外一张表数据导入。
temp文件数据样例：
java zhang web wang java zhao java qin web liu web zheng ios li
linux chen ios yang ios duan linux ma linux xu
java wen web wu
分桶的原理是根据指定的列的计算hash值模余分桶数量后将数据分开存放。方便数据抽样
select * from teacher tablesample(bucket 1 out of 3 on name);
注：分桶语法—TABLESAMPLE(BUCKET x OUT OF y)
y必须是table总bucket数的倍数或者因子。hive根据y的大小，决定抽样的比例。例如：table总共分了3份，当y=3时，抽取(3/3=)1个bucket的数据，当y=6时， 抽取(3/6=)1/2个bucket的数据。
x表示从哪个bucket开始抽取。
例如：table总bucket数为3，tablesample(bucket 3 out of 3)，表示总共抽取
（3/3=）1个bucket的数据，抽取第3个bucket的数据。
再例如：table总bucket数为32，tablesample(bucket 3 out of 16)，表示总共抽取（32/16=）2个bucket的数据，分别为第3个bucket和第（3+16=）19个bucket的数据。
查询第一个桶里数据，并返回一半的数据：
select * from bucketed_user tablesample(bucket 1 out of 6 on id);
Sqoop介绍
sqoop是Apache 提供的工具
用于hdfs和关系型数据库之间数据的导入和导入
可以从hdfs导出数据到关系型数据库，也可以从关系型数据库导入数据到hdfs。
实现步骤：
1.准备sqoop安装包，官网地址：http://sqoop.apache.org
2.配置jdk环境变量和Hadoop的环境变量。因为sqoop在使用是会去找环境变量对应的路径，从而完整工作。
3.sqoop解压即可使用（前提是环境变量都配好了）
4.需要将要连接的数据库的驱动包加入sqoop的lib目录下（本例中用的是mysql数据库） 5.利用指令操作sqoop
（在Sqoop的bin目录下执行下列指令）
说明
指令示例
sh sqoop list-databases --connect jdbc:mysql://192.168.150.138:3306/ -username root -password root
查看指定数据库下的所有表 sh sqoop list-tables --connect jdbc:mysql://hadoop02:3306/hive -username root - password root
实现步骤：
1.现在mysql数据库的test数据下建立一张tabx表，并插入测试数据
sh sqoop import -help（查看import的帮助指令）
建表：create table tabx (id int,name varchar(20));
插入：insert into tabx (id,name) values (1,‘aaa’),(2,‘bbb’),(3,‘ccc’),(1,‘ddd’),(2,‘eee’),(3,‘fff’);
2.进入到sqoop的bin目录下，执行导入语句导入：
sh sqoop import --connect jdbc:mysql://192.168.150.138:3306/test --username root --password root --table tabx --target-dir ‘/sqoop/tabx’ --fields-terminated-by ‘|’ -m 1;
执行：sh sqoop export --connect jdbc:mysql://192.168.150.138:3306/test --username root --password root --export-dir ‘/sqoop/tabx/part-m-00000’ --table taby -m 1 – fields-terminated-by ‘|’
注：sqoop只能导出数据，不能自动建表。所以在导出之前，要现在mysql数据库里建好对应的表
Hive的jdbc编程
hive实现了jdbc接口，所以可以通过java代码操作。但是实际应用中用的不多，一般都是在HDFS储 存的文件基础上建立外部表来进行查询处理。所以jdbc了解一下即可。
实现步骤：
1.在服务器端开启HiveServer服务
./hive --service hiveserver2 & （以后台线程启动）
2.创建本地工程，导入jar包
导入hive\lib目录下的hive-jdbc-1.2.0-standalone.jar
导入hadoop-2.7.1\share\hadoop\common下的hadoop-common-2.7.1.jar
3.编写jdbc代码执行代码示例：
/*
*/ @Test
public void testConnectAndQuery() throws Exception {
//注册数据库驱动，用的hive的jdbc，驱动名固定写死Class.forName(“org.apache.hive.jdbc.HiveDriver”);
//如果用的是hive2服务，则写jdbc:hive2，后面跟上hive服务器的ip以及端口号，端口号默认是10000
Connection conn = DriverManager.getConnection(“jdbc:hive2://192.168.234.21:10000/park”,“root”,“root”
);
Statement stat = conn.createStatement();
ResultSet rs = stat.executeQuery(“select * from stu”);
while(rs.next()){
String name = rs.getString(“name”); System.out.println(name);
}
stat.close();
conn.close();
}
/*
*/ @Test
public void testInsert() throws Exception{
Class.forName(“org.apache.hive.jdbc.HiveDriver”); Connection conn =
DriverManager.getConnection(“jdbc:hive2://192.168.234.21:10000/park”,“root”,“r oot”);
Statement stat = conn.createStatement();
//executeUpdate可用于：创建表,向表中插入数据以及删除表stat.executeUpdate(“insert into table stu values(2,‘rose’)”);
stat.executeUpdate(“create table stu2(id int,name string) row format delimited fields terminated by ’ '”);
stat.executeUpdate(“drop table stu2”);
stat.close();
stat.close();
}
用户接口主要有三个：CLI，JDBC 和 WUI
1.CLI，最常用的模式。实际上在>hive 命令行下操作时，就是利用CLI用户接口。
2.JDBC，通过java代码操作，需要启动hiveserver，然后连接操作。
Metastore
Hive将元数据存储在数据库中，如mysql、derby。Hive中的元数据包括表的名字，
表的列和分区及其属性，表的属性（是否为外部表等），表的数据所在目录等。
解释器（complier)、优化器(optimizer)、执行器(executor)组件
这三个组件用于：HQL语句从词法分析、语法分析、编译、优化以及查询计划的生成。生成的查询计划存储在HDFS中，并在随后有MapReduce调用执行。
Hadoop
Hive的数据存储在HDFS中，大部分的查询、计算由MapReduce完成
1.通过客户端提交一条Hql语句
2.通过complier（编译组件）对Hql进行词法分析、语法分析。在这一步，编译器要知道此hql
语句到底要操作哪张表
3.去元数据库找表信息
4.得到信息
5.complier编译器提交Hql语句分析方案。
6.1executor 执行器收到方案后，执行方案（DDL过程）。在这里注意，执行器在执行方案时，会判断
如果当前方案不涉及到MR组件，比如为表添加分区信息、比如字符串操作等，比如简单的查询操作等，此时就会直接和元数据库交互，然后去HDFS上去找具体数据。
如果方案需要转换成MR job，则会将job 提交给Hadoop的JobTracker。
6.2MR job完成，并且将运行结果写入到HDFS上。
6.3执行器和HDFS交互，获取结果文件信息。
7.如果客户端提交Hql语句是带有查询结果性的，则会发生：7-8-9步，完成结果的查询。
