大数据技术之_08_Hive学习_04_压缩和存储（Hive高级）+ 企业级调优（Hive优化）

[root@hadoop101 software]# tar -zxf jdk-8u144-linux-x64.tar.gz -C /opt/module/
[root@hadoop101 software]# vim /etc/profile
#JAVA_HOME
export JAVA_HOME=/opt/module/jdk1.8.0_144
export PATH=$PATH:$JAVA_HOME/bin
[root@hadoop101 software]# source /etc/profile

[root@hadoop101 software]# tar -zxvf apache-maven-3.0.5-bin.tar.gz -C /opt/module/
[root@hadoop101 software]# vim /etc/profile
#MAVEN_HOME
export MAVEN_HOME=/opt/module/apache-maven-3.0.5
export PATH=$PATH:$MAVEN_HOME/bin
[root@hadoop101 software]# source /etc/profile

[root@hadoop101 software]# yum install svn
[root@hadoop101 software]# yum install autoconf automake libtool cmake
[root@hadoop101 software]# yum install ncurses-devel
[root@hadoop101 software]# yum install openssl-devel
[root@hadoop101 software]# yum install gcc*

[root@hadoop101 software]# tar -zxvf snappy-1.1.3.tar.gz -C /opt/module/
[root@hadoop101 module]# cd snappy-1.1.3/
[root@hadoop101 snappy-1.1.3]# ./configure
[root@hadoop101 snappy-1.1.3]# make
[root@hadoop101 snappy-1.1.3]# make install

# 查看snappy库文件
[root@hadoop101 snappy-1.1.3]# ls -lh /usr/local/lib | grep snappy

[root@hadoop101 software]# tar -zxvf protobuf-2.5.0.tar.gz -C /opt/module/
[root@hadoop101 module]# cd protobuf-2.5.0/
[root@hadoop101 protobuf-2.5.0]# ./configure 
[root@hadoop101 protobuf-2.5.0]# make 
[root@hadoop101 protobuf-2.5.0]# make install

# 查看protobuf版本以测试是否安装成功
[root@hadoop101 protobuf-2.5.0]# protoc --version

[root@hadoop101 software]# tar -zxvf hadoop-2.7.2-src.tar.gz
[root@hadoop101 software]# cd hadoop-2.7.2-src/
[root@hadoop101 software]# mvn clean package -DskipTests -Pdist,native -Dtar -Dsnappy.lib=/usr/local/lib -Dbundle.snappy

hive (default)> set hive.exec.compress.intermediate=true;

hive (default)> set mapreduce.map.output.compress=true;

hive (default)> set mapreduce.map.output.compress.codec=org.apache.hadoop.io.compress.SnappyCodec;

hive (default)> select count(ename) name from emp;

hive (default)> set hive.exec.compress.output=true;

hive (default)> set set hive.exec.compress.output=true;

hive (default)> set mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.SnappyCodec;

hive (default)> set mapreduce.output.fileoutputformat.compress.type=BLOCK;

hive (default)> insert overwrite local directory '/opt/module/datas/snappy-result' select * from emp distribute by deptno sort by empno desc;

[atguigu@hadoop102 snappy-result]$ pwd
/opt/module/datas/snappy-result
[atguigu@hadoop102 snappy-result]$ ll
总用量 4
-rw-r--r--. 1 atguigu atguigu 446 2月  27 22:53 000000_0.snappy

create table log_text(
track_time string,
url string,
session_id string,
referer string,
ip string,
end_user_id string,
city_id string
)
row format delimited fields terminated by '\t'
stored as textfile;

load data local inpath '/opt/module/datas/log.data' into table log_text;

dfs -du -h /user/hive/warehouse/log_text;
+-------------------------------------------------+--+
|                   DFS Output                    |
+-------------------------------------------------+--+
| 18.1 M  /user/hive/warehouse/log_text/log.data  |
+-------------------------------------------------+--+

create table log_orc(
track_time string,
url string,
session_id string,
referer string,
ip string,
end_user_id string,
city_id string
)
row format delimited fields terminated by '\t'
stored as orc;

insert into table log_orc select * from log_text;

dfs -du -h /user/hive/warehouse/log_orc;
+-----------------------------------------------+--+
|                  DFS Output                   |
+-----------------------------------------------+--+
| 2.8 M  /user/hive/warehouse/log_orc/000000_0  |
+-----------------------------------------------+--+

create table log_parquet(
track_time string,
url string,
session_id string,
referer string,
ip string,
end_user_id string,
city_id string
)
row format delimited fields terminated by '\t'
stored as parquet;

insert into table log_parquet select * from log_text;

dfs -du -h /user/hive/warehouse/log_parquet;
+----------------------------------------------------+--+
|                     DFS Output                     |
+----------------------------------------------------+--+
| 13.1 M  /user/hive/warehouse/log_parquet/000000_0  |
+----------------------------------------------------+--+

select count(*) from log_text;
+---------+--+
|   _c0   |
+---------+--+
| 100000  |
+---------+--+
1 row selected (20.145 seconds)

select count(*) from log_orc;
+---------+--+
|   _c0   |
+---------+--+
| 100000  |
+---------+--+
1 row selected (20.44 seconds)

select count(*) from log_parquet;
+---------+--+
|   _c0   |
+---------+--+
| 100000  |
+---------+--+
1 row selected (17.945 seconds)

[atguigu@hadoop102 hadoop-2.7.2]$ hadoop
Usage: hadoop [--config confdir] [COMMAND | CLASSNAME]
  CLASSNAME            run the class named CLASSNAME
 or
  where COMMAND is one of:
  fs                   run a generic filesystem user client
  version              print the version
  jar <jar>            run a jar file
                       note: please use "yarn jar" to launch
                             YARN applications, not this command.
  checknative [-a|-h]  check native hadoop and compression libraries availability
  distcp <srcurl> <desturl> copy file or directories recursively
  archive -archiveName NAME -p <parent path> <src>* <dest> create a hadoop archive
  classpath            prints the class path needed to get the
  credential           interact with credential providers
                       Hadoop jar and the required libraries
  daemonlog            get/set the log level for each daemon
  trace                view and modify Hadoop tracing settings

Most commands print help when invoked w/o parameters.

[atguigu@hadoop102 hadoop-2.7.2]$ hadoop checknative
19/02/27 19:18:38 WARN bzip2.Bzip2Factory: Failed to load/initialize native-bzip2 library system-native, will use pure-Java version
19/02/27 19:18:38 INFO zlib.ZlibFactory: Successfully loaded & initialized native-zlib library
Native library checking:
hadoop:  true /opt/module/hadoop-2.7.2/lib/native/libhadoop.so
zlib:    true /lib64/libz.so.1
snappy:  false 
lz4:     true revision:99
bzip2:   false 
openssl: false Cannot load libcrypto.so (libcrypto.so: 无法打开共享对象文件: 没有那个文件或目录)!

[atguigu@hadoop102 hadoop-snappy]$ pwd
/opt/software/hadoop-snappy
[atguigu@hadoop102 hadoop-snappy]$ tar -zxvf hadoop-2.7.2.tar.gz

[atguigu@hadoop102 native]$ pwd
/opt/software/hadoop-snappy/hadoop-2.7.2/lib/native
[atguigu@hadoop102 native]$ ll
总用量 5188
-rw-r--r--. 1 atguigu atguigu 1210260 9月   1 2017 libhadoop.a
-rw-r--r--. 1 atguigu atguigu 1487268 9月   1 2017 libhadooppipes.a
lrwxrwxrwx. 1 atguigu atguigu      18 2月  27 20:00 libhadoop.so -> libhadoop.so.1.0.0
-rwxr-xr-x. 1 atguigu atguigu  716316 9月   1 2017 libhadoop.so.1.0.0
-rw-r--r--. 1 atguigu atguigu  582048 9月   1 2017 libhadooputils.a
-rw-r--r--. 1 atguigu atguigu  364860 9月   1 2017 libhdfs.a
lrwxrwxrwx. 1 atguigu atguigu      16 2月  27 20:00 libhdfs.so -> libhdfs.so.0.0.0
-rwxr-xr-x. 1 atguigu atguigu  229113 9月   1 2017 libhdfs.so.0.0.0
-rw-r--r--. 1 atguigu atguigu  472950 9月   1 2017 libsnappy.a
-rwxr-xr-x. 1 atguigu atguigu     955 9月   1 2017 libsnappy.la
lrwxrwxrwx. 1 atguigu atguigu      18 2月  27 20:00 libsnappy.so -> libsnappy.so.1.3.0
lrwxrwxrwx. 1 atguigu atguigu      18 2月  27 20:00 libsnappy.so.1 -> libsnappy.so.1.3.0
-rwxr-xr-x. 1 atguigu atguigu  228177 9月   1 2017 libsnappy.so.1.3.0

[atguigu@hadoop102 native]$ cp ../native/* /opt/module/hadoop-2.7.2/lib/native/

[atguigu@hadoop102 lib]$ xsync native/

[atguigu@hadoop102 lib]$ hadoop checknative
19/02/27 20:07:28 WARN bzip2.Bzip2Factory: Failed to load/initialize native-bzip2 library system-native, will use pure-Java version
19/02/27 20:07:28 INFO zlib.ZlibFactory: Successfully loaded & initialized native-zlib library
Native library checking:
hadoop:  true /opt/module/hadoop-2.7.2/lib/native/libhadoop.so
zlib:    true /lib64/libz.so.1
snappy:  true /opt/module/hadoop-2.7.2/lib/native/libsnappy.so.1
lz4:     true revision:99
bzip2:   false 
openssl: false Cannot load libcrypto.so (libcrypto.so: 无法打开共享对象文件: 没有那个文件或目录)!

[atguigu@hadoop102 hadoop-2.7.2]$ sbin/start-dfs.sh -- 启动dfs集群
[atguigu@hadoop103 hadoop-2.7.2]$ sbin/start-yarn.sh -- 启动yarn集群
[atguigu@hadoop102 hive]$ zkstart.sh -- 启动zookeeper集群
[atguigu@hadoop102 hive]$ bin/hive -- 启动hive
[atguigu@hadoop102 hadoop-2.7.2]$ sbin/mr-jobhistory-daemon.sh start historyserver -- 启动历史服务器

create table log_orc_none(
track_time string,
url string,
session_id string,
referer string,
ip string,
end_user_id string,
city_id string
)
row format delimited fields terminated by '\t'
stored as orc tblproperties ("orc.compress"="NONE");

insert into table log_orc_none select * from log_text;

dfs -du -h /user/hive/warehouse/log_orc_none/;
+----------------------------------------------------+--+
|                     DFS Output                     |
+----------------------------------------------------+--+
| 7.7 M  /user/hive/warehouse/log_orc_none/000000_0  |
+----------------------------------------------------+--+

create table log_orc_snappy(
track_time string,
url string,
session_id string,
referer string,
ip string,
end_user_id string,
city_id string
)
row format delimited fields terminated by '\t'
stored as orc tblproperties ("orc.compress"="SNAPPY");

insert into table log_orc_snappy select * from log_text;

dfs -du -h /user/hive/warehouse/log_orc_snappy/;
+------------------------------------------------------+--+
|                      DFS Output                      |
+------------------------------------------------------+--+
| 3.8 M  /user/hive/warehouse/log_orc_snappy/000000_0  |
+------------------------------------------------------+--+

<property>
    <name>hive.fetch.task.conversion</name>
    <value>more</value>
    <description>
      Expects one of [none, minimal, more].
      Some select queries can be converted to single FETCH task minimizing latency.
      Currently the query should be single sourced not having any subquery and should not have
      any aggregations or distincts (which incurs RS), lateral views and joins.
      0. none : disable hive.fetch.task.conversion
      1. minimal : SELECT STAR, FILTER on partition columns, LIMIT only
      2. more  : SELECT, FILTER, LIMIT only (support TABLESAMPLE and virtual columns)
    </description>
</property>

hive (default)> set hive.fetch.task.conversion=none;
hive (default)> select * from emp;
hive (default)> select ename from emp;
hive (default)> select ename from emp limit 3;

set hive.exec.mode.local.auto=true;  // 开启本地mr模式

// 设置local mr的最大输入数据量，当输入数据量小于这个值时采用local mr的方式，默认为134217728，即128M
set hive.exec.mode.local.auto.inputbytes.max=50000000;

// 设置local mr的最大输入文件个数，当输入文件个数小于这个值时采用local mr的方式，默认为4
set hive.exec.mode.local.auto.input.files.max=10;

set hive.exec.mode.local.auto=true;
select count(*) from log_orc;
+---------+--+
|   _c0   |
+---------+--+
| 100000  |
+---------+--+
1 row selected (1.382 seconds)

set hive.exec.mode.local.auto=false;
select count(*) from log_orc;
+---------+--+
|   _c0   |
+---------+--+
| 100000  |
+---------+--+
1 row selected (16.895 seconds)

// 创建大表
create table bigtable(
id bigint, time bigint, uid string, keyword string, url_rank int, click_num int, click_url string
)
row format delimited fields terminated by '\t';

// 创建小表
create table smalltable(
id bigint, time bigint, uid string, keyword string, url_rank int, click_num int, click_url string
)
row format delimited fields terminated by '\t';

// 创建join后表的语句
create table jointable(
id bigint, time bigint, uid string, keyword string, url_rank int, click_num int, click_url string
)
row format delimited fields terminated by '\t';

load data local inpath '/opt/module/datas/bigtable' into table bigtable;
load data local inpath '/opt/module/datas/smalltable' into table smalltable;

set hive.auto.convert.join=false;

insert overwrite table jointable
select b.id, b.time, b.uid, b.keyword, b.url_rank, b.click_num, b.click_url
from smalltable s
left join bigtable b
on b.id = s.id;

No rows affected (31.015 seconds)

insert overwrite table jointable
select b.id, b.time, b.uid, b.keyword, b.url_rank, b.click_num, b.click_url
from bigtable b
left join smalltable s
on s.id = b.id;

No rows affected (32.59 seconds)

<property>
    <name>mapreduce.jobhistory.address</name>
    <value>hadoop102:10020</value>
</property>
<property>
    <name>mapreduce.jobhistory.webapp.address</name>
    <value>hadoop102:19888</value>
</property>

[atguigu@hadoop102 hadoop-2.7.2]$ sbin/mr-jobhistory-daemon.sh start historyserver

// 创建原始表
create table ori(
id bigint, time bigint, uid string, keyword string, url_rank int, click_num int, click_url string
)
row format delimited fields terminated by '\t';

// 创建空id表
create table nullidtable(
id bigint, time bigint, uid string, keyword string, url_rank int, click_num int, click_url string
)
row format delimited fields terminated by '\t';

// 创建join后的表
create table jointable(
id bigint, time bigint, uid string, keyword string, url_rank int, click_num int, click_url string
)
row format delimited fields terminated by '\t';

load data local inpath '/opt/module/datas/ori' into table ori;
load data local inpath '/opt/module/datas/nullid' into table nullidtable;

insert overwrite table jointable 
select n.* from nullidtable n left join ori o on n.id=o.id;

insert overwrite table jointable 
select n.* from (select * from nullidtable where id is not null) n left join ori o on n.id=o.id;

set mapreduce.job.reduces=5;

insert overwrite table jointable
select n.* from nullidtable n left join ori b on n.id=b.id;

set mapreduce.job.reduces=5;

insert overwrite table jointable
select n.* from nullidtable n full join ori o on 
case when n.id is null then concat('hive', rand()) else n.id end=o.id;

set hive.auto.convert.join=true; -- 默认为true

set hive.mapjoin.smalltable.filesize=25000000;

set hive.auto.convert.join=true; -- 默认为true

insert overwrite table jointable
select b.id, b.time, b.uid, b.keyword, b.url_rank, b.click_num, b.click_url
from smalltable s
join bigtable b
on s.id=b.id;

insert overwrite table jointable
select b.id, b.time, b.uid, b.keyword, b.url_rank, b.click_num, b.click_url
from bigtable b
join smalltable s
on s.id=b.id;

set hive.map.aggr=true;

set hive.groupby.mapaggr.checkinterval=100000;

set hive.groupby.skewindata=true;

create table bigtable(
id bigint, time bigint, uid string, keyword string, url_rank int, click_num int, click_url string
) 
row format delimited fields terminated by '\t';

load data local inpath '/opt/module/datas/bigtable' into table bigtable;

set mapreduce.job.reduces=5;

select count(distinct id) from bigtable;

+--------+--+
|   c0   |
+--------+--+
| 99947  |
+--------+--+
1 row selected (46.446 seconds)

select count(id) from (select id from bigtable group by id) a;

+--------+--+
|  _c0   |
+--------+--+
| 99947  |
+--------+--+
1 row selected (78.134 seconds)

select o.id from bigtable b
join ori o on o.id=b.id
where o.id<=10;

100 rows selected (23.533 seconds)

select b.id from bigtable b
join (select id from ori where id<=10) o on b.id=o.id;

100 rows selected (22.041 seconds)

set hive.exec.dynamic.partition=true;

set hive.exec.dynamic.partition.mode=nonstrict;

set hive.exec.max.dynamic.partitions=1000;

set hive.exec.max.dynamic.partitions.pernode=100;

set hive.exec.max.created.files=100000;

set hive.error.on.empty.partition=false;

create table ori_partitioned(
id bigint, time bigint, uid string, keyword string, url_rank int, click_num int, click_url string
) 
partitioned by(p_time bigint) 
row format delimited fields terminated by '\t';

load data local inpath '/opt/module/datas/ds1' into table ori_partitioned partition(p_time='20111230000010');
load data local inpath '/opt/module/datas/ds2' into table ori_partitioned partition(p_time='20111230000011');

create table ori_partitioned_target(
id bigint, time bigint, uid string, keyword string, url_rank int, click_num int, click_url string
)
partitioned by(p_time string)
row format delimited fields terminated by '\t';

set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.exec.max.dynamic.partitions=1000;
set hive.exec.max.dynamic.partitions.pernode=100;
set hive.exec.max.created.files=100000;
set hive.error.on.empty.partition=false;

insert overwrite table ori_partitioned_target partition(p_time) 
select id, time, uid, keyword, url_rank, click_num, click_url, p_time from ori_partitioned;

show partitions ori_partitioned_target;

create table if not exists default.emp(
empno int,
ename string,
job string,
mgr int,
hiredate string, 
sal double, 
comm double,
deptno int
)
row format delimited fields terminated by '\t';

load data local inpath '/opt/module/datas/emp.txt' into table default.emp;

create table if not exists default.emp_partiton1(
empno int,
ename string,
job string,
mgr int,
hiredate string, 
sal double, 
comm double
)
partitioned by(deptnoooooo int)
row format delimited fields terminated by '\t';

set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.exec.max.dynamic.partitions=1000;
set hive.exec.max.dynamic.partitions.pernode=100;
set hive.exec.max.created.files=100000;
set hive.error.on.empty.partition=false;

insert overwrite table emp_partiton1 partition(deptnoooooo) 
select empno, ename, job, mgr, hiredate, sal, comm, deptno from emp;

show partitions emp_partiton1;

+-----------------+--+
|    partition    |
+-----------------+--+
| deptnoooooo=10  |
| deptnoooooo=20  |
| deptnoooooo=30  |
+-----------------+--+

set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

set mapreduce.job.reduces=-1; -- 设置mapreduce的个数

select count(*) from emp;
Hadoop job information for Stage-1: number of mappers: 1; number of reducers: 1

set mapreduce.input.fileinputformat.split.maxsize;
+----------------------------------------------------------+--+
|                           set                            |
+----------------------------------------------------------+--+
| mapreduce.input.fileinputformat.split.maxsize=256000000  |
+----------------------------------------------------------+--+

set mapreduce.input.fileinputformat.split.maxsize=100;

select count(*) from emp;
Hadoop job information for Stage-1: number of mappers: 6; number of reducers: 1

hive.exec.reducers.bytes.per.reducer=256000000

hive.exec.reducers.max=1009

N=min(参数2，总输入数据量/参数1)

set mapreduce.job.reduces=15;

set hive.exec.parallel=true;              // 打开任务并行执行
set hive.exec.parallel.thread.number=16;  // 同一个sql允许最大并行度，默认为8

<property>
    <name>hive.mapred.mode</name>
    <value>strict</value>
    <description>
      The mode in which the Hive operations are being performed. 
      In strict mode, some risky queries are not allowed to run. They include:
        Cartesian Product.
        No partition being picked up for a query.
        Comparing bigints and strings.
        Comparing bigints and doubles.
        Orderby without limit.
    </description>
</property>

select * from emp_partiton;

select * from emp order by sal;

select e.ename, d.dname from emp e, dept d;

<property>
  <name>mapreduce.job.jvm.numtasks</name>
  <value>10</value>
  <description>How many tasks to run per jvm. If set to -1, there is
  no limit. 
  </description>
</property>

<property>
    <name>mapreduce.map.speculative</name>
    <value>true</value>
    <description>If true, then multiple instances of some map tasks may be executed in parallel.</description>
</property>

<property>
    <name>mapreduce.reduce.speculative</name>
    <value>true</value>
    <description>If true, then multiple instances of some reduce tasks may be executed in parallel.</description>
</property>

<property>
    <name>hive.mapred.reduce.tasks.speculative.execution</name>
    <value>true</value>
    <description>Whether speculative execution for reducers should be turned on. </description>
</property>

EXPLAIN [EXTENDED | DEPENDENCY | AUTHORIZATION] query

hive (default)> explain select * from emp;
hive (default)> explain select deptno, avg(sal) avg_sal from emp group by deptno;

hive (default)> explain extended select * from emp;
hive (default)> explain extended select deptno, avg(sal) avg_sal from emp group by deptno;