Hive+Sqoop浅度学习指南

摘要：业务需求统计每小时的数数据采集数据清洗用来描述将数据从来源端经过抽取转换加载至目的端的过程字段过滤字段补全用户信息商品信息字段格式化数据分析将数据导出介绍由开源的用于解决海量结构化日志的数据统计的项目本质将转化为程序的其实时上的目录和

业务

需求：统计每小时的PV数

数据采集

hdfs

hive

数据清洗（ETL）

用来描述将数据从来源端经过抽取（extract）、转换（transform）、加载（load）至目的端的过程

字段过滤

"31/Aug/2015:00:04:37 +0800"

"GET /course/view.php?id=27 HTTP/1.1"

字段补全

用户信息、商品信息-》RDBMS

字段格式化

2015-08-31 00:04:37 20150831000437

数据分析

MapReduce

Hive

Spark

将数据导出

​

hive介绍

由Facebook开源的,用于解决海量结构化日志的数据统计的项目

本质: 将HQL转化为MapReduce程序

Hive的其实时HDFS上的目录和文件

​

Hive的安装模式

嵌入模式

元数据信息被保存在自带的Deybe数据库中

只允许创建一个连接

多用于Demo

本地模式

元数据信息被保存在MySQL数据库

MySQL数据库与Hive运行在同一台物理机器上

多用于开发和测试

远程模式

元数据信息被保存在MySQL数据库

MySQL数据库与Hive运行在不同台物理机器上

用于实际生成环境

Linux下MySQL安装

​ 1) 卸载

​ $ rpm -qa | grep mysql

​ $ sudo rpm -e mysql-libs-5.1.71-1.el6.x86_64 --nodeps

​ 2) 安装

​ 可选择将缓存替换,然后再安装 $ sudo cp -r /opt/software/x86_64/ /var/cache/yum/

​ $ sudo yum install -y mysql-server mysql mysql-devel

​ 3) 启动mysql服务

​ $ sudo service mysqld start

​ 4) 设置密码

​ $ /usr/bin/mysqladmin -u root password "新密码"

​ 5) 开机启动

​ $ sudo chkconfig mysqld on

​ 6) 授权root的权限及设置远程登录

​ 登录

​ $ mysql -u root -p

​ 授权

mysql> grant all privileges on *.* to "root"@"%" identified by "密码";
mysql> grant all privileges on *.* to "root"@"linux01" identified by "密码"; -- 必须有这一句，%包括所有

​ all privileges 所有权限

​ . 所有数据库的所有表

​ "root"@"%" 在任意主机以root身份登录

​ "root"@"linux03.ibf.com" 在linux03主机以root登录

​ by "root" 使用root作为密码

​ 7)刷新授权

mysql> flush privileges;

​ 8)测试,在windows中是否可以登录

 mysql -h linux03.ibf.com -u root -p

hive环境搭建:本地模式

​ 必须先安装HDFS和Yarn

1)安装:

​ $ tar -zxvf /opt/software/hive-0.13.1-bin.tar.gz -C /opt/modules/

​ 重命名hive文件夹名字

​ $ cd /opt/modules

​ $ mv apache-hive-0.13.1-bin/ hive-0.13.1/

2)在HDFS上 创建tmp目录和hive仓库

​ $ bin/hdfs dfs -mkdir -p /user/hive/warehouse

​ $ bin/hdfs dfs -mkdir /tmp #已存在

​ $ bin/hdfs dfs -chmod g+w /user/hive/warehouse

​ $ bin/hdfs dfs -chmod g+w /tmp

3)修改配置

​ $ cd hive-0.13.1/

​ $ cp conf/hive-default.xml.template conf/hive-site.xml

​ $ cp conf/hive-log4j.properties.template conf/hive-log4j.properties

​ $ cp conf/hive-env.sh.template conf/hive-env.sh

​ 3-1)修改hive-env.sh

JAVA_HOME=/opt/modules/jdk1.7.0_67 #添加

HADOOP_HOME=/opt/modules/hadoop-2.5.0

export HIVE_CONF_DIR=/opt/modules/hive-0.13.1/conf

​ 3-2)修改hive.site.xml

​

    javax.jdo.option.ConnectionURL
    jdbc:mysql://linux01:3306/metastore?createDatabaseIfNotExist=true


    javax.jdo.option.ConnectionDriverName
    com.mysql.jdbc.Driver


    javax.jdo.option.ConnectionUserName
    root


    javax.jdo.option.ConnectionPassword
    123456

​ 3-3)修改日志配置hive-log4j.properties

​ hive.log.dir=/opt/modules/hive-0.13.1/logs

​ 3-4)拷贝jdbc驱动到hive的lib目录

​ $ cp /opt/software/mysql-connector-java-5.1.34-bin.jar /opt/modules/hive-0.13.1/lib/

​ 4)确定yarn和hdfs启动

​ $ jps

​ 6468 ResourceManager

​ 6911 Jps

​ 6300 RunJar

​ 6757 NodeManager

​ 2029 NameNode

​ 2153 DataNode

​ 此时使用bin/hive 可以进入hive

hive启动及基本使用

​ 进入hive目录

​ $ cd /opt/modules/hive-0.13.1/

进入hive

​ bin/hive

基本命令

​ show databases;

​ create database mydb;

​ use mydb;

​ show tables;

创建表及加载数据

create table student (
id int comment "id of student",
name string comment "name of student",
age int comment "age of student",
gender string comment "sex of student",
addr string
)    
comment "this is a demo"
row format delimited fields terminated by "	";

表默认创建在/user/hive/warehouse里

通过hive.metastore.warhouse.dir配置

查看表

desc student; 查看表字段

或

desc formatted student; 可以查看元数据

此时mysql的metastore数据库状况

mysql> select * from TBLS;

+--------+-------------+-------+------------------+-------+-----------+-------+----------+---------------+--------------------+--------------------+
| TBL_ID | CREATE_TIME | DB_ID | LAST_ACCESS_TIME | OWNER | RETENTION | SD_ID | TBL_NAME | TBL_TYPE      | VIEW_EXPANDED_TEXT | VIEW_ORIGINAL_TEXT |
+--------+-------------+-------+------------------+-------+-----------+-------+----------+---------------+--------------------+--------------------+
|      1 |  1556132119 |     6 |                0 | chen  |         0 |     1 | student  | MANAGED_TABLE | NULL               | NULL               |
+--------+-------------+-------+------------------+-------+-----------+-------+----------+---------------+--------------------+--------------------+
1 row in set (0.00 sec)

mysql> select * from COLUMNS_V2;

+-------+-----------------+-------------+-----------+-------------+
| CD_ID | COMMENT         | COLUMN_NAME | TYPE_NAME | INTEGER_IDX |
+-------+-----------------+-------------+-----------+-------------+
|     1 | NULL            | addr        | string    |           4 |
|     1 | age of student  | age         | int       |           2 |
|     1 | sex of student  | gender      | string    |           3 |
|     1 | id of student   | id          | int       |           0 |
|     1 | name of student | name        | string    |           1 |
+-------+-----------------+-------------+-----------+-------------+
5 rows in set (0.00 sec)

加载数据(在家目录下创建student.log)

load data local inpath "/home/hadoop/student.log" into table student;

从hdfs上加载( 加载完,hdfs上的student.data到 表目录下)

load data inpath "/input/student.data" into table student;

在命令行内设置配置

​ 重启无效

​ set hive.cli.print.header=true; #列名

​ set hive.cli.print.current.db=true; #表名

​ reset; 重置

​ 重启有效

    

​    hive.cli.print.header

​    true





​    hive.cli.print.current.db

​    true

与Linux交互

!ls

!pwd

与hadoop交互

dfs -ls /

dfs -mkdir /hive

hive的脚本

-e 执行sql

-f 执行sql文件

-S 静默执行

hive -e

$ bin/hive -e "select *from test_db.emp_p"

hive -f

$ bin/hive -S -f /home/hadoop/emp.sql > ~/result.txt

删除表

drop table user;

清空表

truncate table user;

表类型 内部表（管理表 MANAGED_TABLE）

create table emp(
empId int,
empString string,
job string,
salary float,
deptId int
)
row format delimited fields terminated by "	";

load data inpath "/input/dept.txt" into table dept;

# 或从本地加载 
load data local inpath "/home/hadoop/dept.txt" into table dept;

外部表（EXTERNAL_TABLE ）

create external table emp_ex (
empId int,
empName string,
job string,
salary float,
deptId int
)
row format delimited fields terminated by "	"
location "/hive/table/emp";

把数据移动到表所在位置

hive (mydb)> dfs -mv /input/emp.txt /hive/table/emp/emp.txt 

服务器加载

hive (mydb)> load data local inpath "/home/hadoop/emp.data" into table emp;

或者直接使用dfs命令移动数据到hive表目录下

hive (mydb)> dfs -put /home/hadoop/emp.data  /hello/table/emp;

内部表和外部表区别

创建表

外部表创建表的时候，需要用external

删除表

外部表在删除表的时候只会删除表的元数据(metadata)信息不会删除表数据(data)

内部表删除时会将元数据信息和表数据同时删除

内部表数据由Hive自身管理，外部表数据由HDFS管理

内部表数据存储的位置是hive.metastore.warehouse.dir（默认：/user/hive/warehouse），外部表数据的存储位置由自己制定；

分区表 创建分区表

create table emp_part(
empno int,
empname string,
empjob string,
mgrno int,
birthday string,
salary float,
bonus float,
deptno  int
)
partitioned by (province string)
row format delimited fields terminated by "	";

向分区表加载数据

显式指定分区值

load data local inpath "/home/user01/emp.txt" into table emp_part partition (province="CHICAGO");

分区操作 查看分区

show partitions emp_part;

添加分区

alter table emp_part add partition (province="shanghai");

删除分区

alter table emp_part drop partition (province="shanghai");

向分区添加数据

load data local inpath "本地路径" into table emp_part partition (province="shanghai");

查询分区数据

select * from emp_part where province="henan";

二级分区 创建二级分区

create table emp_second(
id int ,
name string,
job string,
salary float,
dept int
)
partitioned by (day string,hour string)
row format delimited fields terminated by "	";

添加分区

alter table emp_second add partition (day="20180125",hour="16");

删除分区

alter table emp_second drop partition (day="20180125");

添加数据的时候指定分区(没有该分区会创建)

load data local inpath "/home/hadoop/emp.log" into table emp_second partition (day="20180125",hour="17");

桶表

连接两个在相同列上划分了桶的表，使用map side join 实现

使sampling更高效

需设置set hive.enforce.bucketing=true

create table bucketed_users(id int, name string)
clustered by (id) into 4 buckets

某个数据被分到哪个桶根据指定列的hash值对桶数取余得到

导入方式总结 本地导入

load data local inpath "本地路径" into table 表名

bin/hdfs dfs -put 本地路径 hdfs路径(hive的表位置)

hdfs上导入

load data inpath "hdfs路径" into table 表名

覆盖写

load data inpath "hdfs路径" overwrite into table 表名

load data local inpath "本地路径" overwrite into table 表名

通过insert语句将select的结果 插入到一张表中

insert into table test_tb select * from emp_p;

创建表时加载数据

create external table test_tb (

    id int,

    name string

)

row format delimited fields terminated by "	";

location "/hive/test_tb";

sqoop方式 导出方式 hive 脚本 hive -e

bin/hive -e "use test_db;select * from emp_p" > /home/hadoop/result.txt

hive -f 执行sql文件

bin/hive -f 路径 >> /home/hadoop/result.txt

hive>

导出到本地 (默认分隔符是 ASSII 001)

insert overwrite local directory "/home/hadoop/data" select * from emp_p;

insert overwrite local directory "/home/hadoop/data" row format delimited fields terminated by "^" select * from emp_p;

到出HDFS

hive > insert overwrite directory "/data" select * from emp_p;

export 和 import (HDFS)

hive > export table emp_p to "/input/export" ;

hive > import table emp_imp from "hdfs_path" ;

HQL

http://hive.apache.org/

常用语法

通配 *指定字段

select id,name from emp;

where 条件查询

select * from emp_p where salary > 10000;

between and

select * from emp_p where sal between 10000 and 15000;

is null| is not null

select * from user where email is not null;

in () | not in ()

select * from emp_p where did in (1,2,3);

聚合函数

count max min sum avg

select count(1) personOfDept from emp_p group by job;

select sum(sal) from emp_p;

distinct

select distinct id from emp_part;

select distinct name, province from emp_part;

子查询

select eid,ename,salary ,did from emp where emp.did in (select did from dept where dname="人事部");

表连接

emp.eid emp.ename       emp.salary      emp.did
1001    jack    10000.0 1
1002    tom     2000.0  2
1003    lily    20000.0 3
1004    aobama  10000.0 5
1005    yang    10000.0 6

dept.did        dept.dname      dept.dtel
1       人事部  021-456
2       财务部  021-234
3       技术部  021-345
4       BI部    021-31
5       产品部  021-232

积

select * from dept, emp;

select * from emp, dept where emp.did=dept.did;

join

select t1.eid, t1.ename, t1.salary,t2.did ,t2.dname from emp t1 join dept t2 on t1.did=t2.did;

外连查询

left join

select eid,ename, salary,t2.did, t2.dname from emp t1 left join dept t2 on t1.did = t2.did;

right join

select eid,ename, salary,t2.did, t2.dname from emp t1 right join dept t2 on t1.did = t2.did;

全连接

select eid,ename, salary,t2.did, t2.dname from emp t1 full join dept t2 on t1.did = t2.did;

hql的四种排序 全局排序Order By (对所有的数据进行排序)

select * from emp_part order by salary;

设置reduce个数为3,也只有一个文件

set mapreduce.job.reduces=3;

内部排序sort by (每个reduce内部进行排序 )

底层 时在reduce函数之前完成的

设置reduce个数

set mapreduce.job.reduces=2;

insert overwrite local directory "/home/hadoop/result" select * from emp_part sort by salary; # 默认reduce个数为1, 这种情况下和order by一样

分区排序（通过distribute by设置分区 ，使用 sort by设置分区内排序）

set mapreduce.job.reduces=3;

这里使用部分分区,薪资排序

insert overwrite local directory "/home/hadoop/result" select * from emp_part distribute by deptno sort by salary;

Cluster By (distribute by 和sort by条件一致时 使用cluster by) 连接方式 配置

修改hive-site.xml

hive.server2.long.polling.timeout
5000



hive.server2.thrift.port
10000



hive.server2.thrift.bind.host
bigdata.ibf.com

mysql数据库中创建一个普通用户

1)创建用户

 CREATE USER "hadoop"@"centos01.bigdata.com" IDENTIFIED BY "123456";

2)授权访问（hive的存储元数据的数据库）

GRANT ALL ON metastore.* TO "hadoop"@"centos01.bigdata.com" IDENTIFIED BY "123456";  

GRANT ALL ON metastore.* TO "hadoop"@"%" IDENTIFIED BY "123456";

3)刷新授权

flush privileges;

beeline

启动服务

$ bin/hiveserver2 &
或
$bin/hive --service hiveserver2 &

连接

$ bin/beeline
beeline>!connect jdbc:hive2://bigdata.ibf.com:10000
输入mysql的用户名
输入mysql密码

sqoop的介绍及安装部署

功能：用于HDFS与RDBMS之间数据的导入导出

所有的导入导出都是基于HDFS而言

数据分析流程

数据采集
                     日志;    
                     RDBMS; 使用sqoop,将需要分析的数据采集到HDFS
数据清洗
            字段过滤
           字段补全            -》将需要分析的字段导入到HDFS
           字段格式化               
数据分析
            将分析后的数据存储在HDFS
            将结果数据从HDFS导出到MySQL
数据展示
                     从RDBMS中读取数据

sqoop支持：HDFS，hive，hbase

sqoop的底层

    -》使用sqoop命令，通过不同的参数，实现不同的需求

    -》sqoop根据不同的参数，解析后传递给底层的MapReduce模板

    -》将封装好的MapReduce打成jar包，提交给yarn执行

    -》这个MapReduce只有maptask，没有reducetask

版本

    -》sqoop1

    -》sqoop2：

        -》多了server端

        -》添加了安全机制

安装部署

下载解压

tar -zxvf /opt/software/sqoop-1.4.5-cdh5.3.6.tar.gz  -C /opt/cdh-5.3.6/

修改配置文件

$ pwd
/opt/cdh-5.3.6/sqoop-1.4.5-cdh5.3.6
$ cp conf/sqoop-env-template.sh conf/sqoop-env.sh

修改sqoop-env.sh

 export HADOOP_COMMON_HOME=/opt/cdh-5.3.6/hadoop-2.5.0-cdh5.3.6
 #Set path to where hadoop-*-core.jar is available
 export HADOOP_MAPRED_HOME=/opt/cdh-5.3.6/hadoop-2.5.0-cdh5.3.6
 #Set the path to where bin/hive is available
 export HIVE_HOME=/opt/cdh-5.3.6/hive-0.13.1-cdh5.3.6

将MySQL连接驱动放入sqoop的lib目录

$ cp /opt/software/mysql-connector-java-5.1.34-bin.jar  /opt/cdh-5.3.6/sqoop-1.4.5-cdh5.3.6/lib/

使用测试

查看命令信息

[hadoop@linux03 sqoop-1.4.5-cdh5.3.6]$ bin/sqoop help

查看数据库

[hadoop@linux03 sqoop-1.4.5-cdh5.3.6]$ bin/sqoop list-databases 
--connect jdbc:mysql://linux03.ibf.com:3306 
--username root 
--password 123456

解决sqoop1.4.6-cdh-5.14.2的报错问题

在sqoop-1.4.6中,需要添加java-json包

$ cp /opt/software/java-json.jar /opt/cdh5.14.2/sqoop-1.4.6-cdh5.14.2/lib/

解决找不到hive仓库的问题

$ cp ${HIVE_HOME}/conf/hive-site.xml ${SQOOP_HOME}/conf/

在HADOOP_CLASSPATH中追加hive的依赖

$ sudo vi /etc/profile
#HADOOP_CLASSPATH
export HADOOP_CLASSPATH=$HADOOP_CLASSPATH:/opt/cdh5.14.2/hive-1.1.0-cdh5.14.2/lib/*
source /etc/profile

​

sqoop的导入

bin/sqoop import --help #查看命令提示

导入到HDFS

源：MySQL的一张表

目标：HDFS一个路径

在MySQL中创建测试表

在mysql中添加数据

use test_db;
create table user(
id int primary key,
name varchar(20) not null,
salary float
)charset=utf8;
insert into user values(1,"张三",9000);
insert into user values(2,"李四",10000);
insert into user values(3,"王五",6000);

案例一：mysql-> hdfs (导入到默认路径)

把mysql 中test_db.user 导入到HDFS上, 默认在hdfs://linux01:8020/user/hadoop/

[hadoop@linux03 sqoop-1.4.5-cdh5.3.6]$ bin/sqoop import 
> --connect jdbc:mysql://linux03.ibf.com:3306/mydb 
> --username root 
> --password 123456 
> --table user

当没有reduce时, 有几个map就有几个输出文件

案例二：mysql-> hdfs制定路径及map个数

​ -》指定hdfs输出目录：--target-dir
​ -》指定map的个数：-m

[hadoop@linux03 sqoop-1.4.5-cdh5.3.6]$ bin/sqoop import 
> --connect jdbc:mysql://linux03.ibf.com:3306/test_db 
> --username root 
> --password root 
> --table user 
> --target-dir /toHdfs 
> -m 1

案例三：

​ -》修改导出分隔符 --fields-terminated-by
​ -》--direct 导入更快
​ -》提前删除输出目录

[hadoop@linux03 sqoop-1.4.5-cdh5.3.6]$ bin/sqoop import 
> --connect jdbc:mysql://linux03.ibf.com:3306/test_db 
> --username root 
> --password root 
> --table toHdfs 
> --target-dir /toHdfs 
> --direct 
> --delete-target-dir 
> --fields-terminated-by "	" 
> -m 1

案例四：导入指定的列：--columns

[hadoop@linux03 sqoop-1.4.5-cdh5.3.6]$ bin/sqoop import 
> --connect jdbc:mysql://linux03.ibf.com:3306/mydb 
> --username root 
> --password 123456 
> --table user 
> --columns name,salary 
> --fields-terminated-by "-" 
> --target-dir /sqoop 
> --delete-target-dir 
> --direct 
> -m 1

将SQL语句执行的结果进行导入-e,--query

bin/sqoop import 
--connect jdbc:mysql://bigdata01.com:3306/test 
--username root 
--password 123456 
-e "select * from user where salary>9000 and $CONDITIONS" 
--target-dir /toHdfs 
--delete-target-dir 
-m 1

在上面的-e的查询语句中必须包含where $CONDITIONS ,

    如果想用where语句
     where salary>9000 and $CONDITIONS"

可以设置密码文件(把--password 改为 --password-file)

sqoop会读取整个password-file，包括空格和回车，可以使用echo -n命令生成密码文件，如：echo -n "secret" > password.file

$ echo -n "root" > /home/hadoop/mysqlpasswd && chmod 400 /home/hadoop/mysqlpasswd
bin/sqoop import 
--connect jdbc:mysql://bigdata01.com:3306/test 
--username root 
--password-file  file:///home/hadoop/mysqlpasswd 
-e "select * from toHdfs where $CONDITIONS" 
--target-dir /sqoop 
--delete-target-dir 
-m 1

导入到HIVE

hive,指定数据库中没有该表, 就会创建该表

bin/sqoop import 
--connect jdbc:mysql://linux03.ibf.com:3306/mydb 
--username root 
-P 
--table user 
--fields-terminated-by "	" 
--delete-target-dir 
-m 1 
--hive-import 
--hive-database test_db 
--hive-table user

增量导入

  过程
            MapReduce将数据导入到hdfs用户的家目录
            从家目录将数据导入到hive表
        增量导入
            追加：根据某一列上一次导入的最后一个值，来判断追加的数据
            时间戳：根据数据记录修改的时间戳来进行导入
      --check-column         Source column to check for incremental   change
      --incremental     Define an incremental import of type   "append" or "lastmodified"
      --last-value           Last imported value in the incremental  check column

如果HDFS上没有该文件会创建该文件

bin/sqoop import 
--connect jdbc:mysql://linux03.ibf.com:3306/mydb 
--username root 
--password 123456 
--table user 
--fields-terminated-by "	" 
--target-dir /sqoop/incremental 
-m 1 
--direct 
--check-column id 
--incremental append 
--last-value 3

sqoop job

创建sqoop job,自动创建增量 (报错)

Sqoop job相关的命令有两个：

bin/sqoop job

bin/sqoop-job

使用这两个都可以

创建job：--create

删除job：--delete

执行job：--exec

显示job：--show

列出job：--list

创建job

bin/sqoop-job 
--create your-sync-job 
-- import 
--connect jdbc:mysql://linux03.ibf.com:3306/mydb 
--username root 
-P 
--table user 
-m 1 
--target-dir /hive/incremental 
--incremental append 
--check-column id 
--last-value 1 

查看job

bin/sqoop-job --show your-sync-job

bin/sqoop job --show your-sync-job
bin/sqoop job --exec your-sync-job

bin/sqoop job --list
bin/sqoop job --delete my-sync-job

​

sqoop的导出

将数据从hive(HDFS上的文件与目录),HDFS导出到MySQL

use mydb
create table user_export(
id int primary key,
name varchar(20) not null,
salary float
); 

需要现在数据库中建立表

bin/sqoop export 
--connect jdbc:mysql://linux03.ibf.com:3306/mydb 
--username root 
-P 
--table user_export 
--export-dir /hive/incremental 
--input-fields-terminated-by "," 
-m 1  

使用sqoop --options-file

创建文件 vi sqoopScript ( 所有参数一行一个)

编辑文件sqoopScript

export
--connect 
jdbc:mysql://linux03.ibf.com:3306/test_db
--username
root
-P
--table
emp
-m
1
--export-dir
/input/export
--fields-terminated-by
"	"

执行导出()

bin/sqoop --options-file ~/sqoopScript

内容

Hive简单案例需求分析及结果的导出

动态分区的介绍及使用

使用脚本动态加载到hive表中

hive函数

简单日志流量案例

1 需求及分析

需求

分析统计每天每小时的PV数和UV数

分析

创建数据源表

​ 创建分区表(天,小时)/ 加载数据

数据清洗

​ 创建hive表

​ 字段过滤

​ id url guid 字段补全(无) 字段格式化(无)

数据分析

​ pv：count(url) uv：count(distinct guid)

保存结果

​ 日期(天) 小时 PV UV

导出结果

​ 导出到MySQL

2 具体实现

数据原表

1) 创建原表

create database if not exists hive_db;

user hive_db;

create table tracklogs(

id string,

url string,

referer string,

keyword string,

type string,

guid string,

pageId string,

moduleId string,

linkId string,

attachedInfo string,

sessionId string,

trackerU string,

trackerType string,

ip string,

trackerSrc string,

cookie string,

orderCode string,

trackTime string,

endUserId string,

firstLink string,

sessionViewNo string,

productId string,

curMerchantId string,

provinceId string,

cityId string,

fee string,

edmActivity string,

edmEmail string,

edmJobId string,

ieVersion string,

platform string,

internalKeyword string,

resultSum string,

currentPage string,

linkPosition string,

buttonPosition string

)

partitioned by (date string,hour string)

row format delimited fields terminated by "t";

2) 加载数据

load data local inpath "/opt/datas/2015082818" into table tracklogs partition(date="20150828",hour="18");

load data local inpath "/opt/datas/2015082819" into table tracklogs partition(date="20150828",hour="19");

分析

1) 建立数据分析表

create table clear (

id string,

url string,

guid string

)

partitioned by (date string, hour string)

row format delimited fields terminated by "t";

2) 过滤数据

insert into table clear partition(date="20150828",hour="18") select id,url,guid from tracklogs where date="20150828" and hour="18";

insert into table clear partition(date="20150828",hour="19") select id,url,guid from tracklogs where date="20150828" and hour="19";

3) 指标分析

pv : select date,hour,count(url) as pv from clear group by date,hour;

uv: select date,hour, count(distinct guid) as uv from clear group by date,hour;

保存结果到result

create table result as select date,hour, count(url) pv, count(distinct guid) as uv from clear group by date,hour;

创建表时没指定分隔符则默认分隔符为 001
导出结果到mysql

# 创建表

create table result(

day varchar(30),

hour varchar(30),

pv varchar(30) not null,

uv varchar(30) not null,

primary key(day,hour)

);

# 导出数据

[hadoop@linux03 sqoop-1.4.5-cdh5.3.6]$ bin/sqoop export

--connect jdbc:mysql://linux03.ibf.com:3306/mydb

--username root

--password root

--table result

--export-dir /user/hive/warehouse/hive_db.db/result

--input-fields-terminated-by "001"

-m 1

动态分区表

开启动态分区

set hive.exec.dynamic.partition=true;

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

打开动态分区后，动态分区的模式，有 strict和 nonstrict 两个值可选，strict 要求至少包含一个静态分区列，nonstrict则无此要求。

创建表

create table clear_dynamic (

id string,

url string,

guid string

)

partitioned by (date string, hour string)

row format delimited fields terminated by "t";

动态加载数据

直接加载20180129的所有hour的数据

insert into table clear_dynamic partition(date="20180129",hour) select id,url,guid,hour from tracklogs where date="20180129";

根据hour自动分区

以前是这样写

insert into table clear partition(date="20150828",hour="18") select id,url,guid from tracklogs where date="20150828" and hour="18";

insert into table clear partition(date="20150828",hour="19") select id,url,guid from tracklogs where date="20150828" and hour="19";

使用脚本动态加载到hive表中

20180129/

​ 2018012900

​ 2018012901

​ 2018012902

​ 2018012903

​ 2018012904

​ 2018012905

1) 编写shell_脚本(bin/hive -e "" )

2) 测试脚本

show partitions tracklogs; #查看分区

alter table tracklogs drop partition(date="20150828",hour="18"); 删除分区

alter table tracklogs drop partition(date="20150828",hour="19");

select count(1) from tracklogs; #查看记录数

3) 使用shell脚本使用(bin/hive -f )

4)测试

show partitions tracklogs; #查看分区

alter table tracklogs drop partition(date="20150828",hour="18"); 删除分区

alter table tracklogs drop partition(date="20150828",hour="19");

select count(1) from tracklogs; #查看记录数

Hive函数

用户自定义函数,用于实现hive中不能实现的业务逻辑处理

类型:

​ UDF: 一进一出

​ UDAF: 多进一出 sum,count等

​ UDTF: 一进多出 行列转换

编写UDF:

​ 编写UDF必须继承UDF

​ 必须至少实现一个evaluale方法

​ 必须要有返回类型,可以是null

​ 建议使用hadoop序列化类型

需求:日期转换

​ 31/Aug/2015:00:04:37 +0800 --> 2015-08-31 00:04:37

实现步骤

​ 1) 自定义类实现UDF类

​ 2) 打包不要指定主类

​ 3) 添加到hive中

maven中导入hadoop的包和hive的包

    org.apache.hive
    hive-exec
    1.2.2

具体实现范例

package com.myudf;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Locale;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.Text;

public class DateFormate extends UDF {
    SimpleDateFormat inputDate = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss",Locale.ENGLISH);
    SimpleDateFormat outDate = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    
    //   31/Aug/2015:00:04:37 +0800 -->  2015-08-31 00:04:37
    public Text evaluate(Text str) {
        if(str == null) {
            return null;
        }
        if(StringUtils.isBlank(str.toString())) {
            return null;
        }
        Date date = null;
        String val = null;
        try {
            date = inputDate.parse(str.toString());
            val = outDate.format(date);
        } catch (ParseException e) {
            e.printStackTrace();
        }
        return new Text(val);
    }
    
    public static void main(String[] args) {
        Text val = new DateFormate().evaluate(new Text("31/Aug/2015:00:04:37 +0800"));
        System.out.println(val);
    }
}

hive (test_db)>add jar /home/hadoop/DDD.jar;

hive (test_db)> CREATE TEMPORARY FUNCTION removequote as "com.myudf.date.RemoveQuoteUDF";

hive (test_db)> show functions;

hive压缩格式 压缩格式

bzip2, gzip, lzo, snappy等

压缩比：bzip2>gzip>lzo bzip2

压缩解压速度：lzo>gzip>bzip2 lzo

hadoop支持的压缩格式

bin/hadoop checknative  -a

http://google.github.io/snappy/

配置压缩

编译hadoop源码：

mvn package -Pdist,native,docs -DskipTests -Dtar  -Drequire.snappy

替换$HADOOP_HOME/lib/native

关闭hadoop相关进程

解压cdh5.xxx-snappy-lib-native.tar.gz 到$HADOOP_HOME/lib

$ tar -zxvf native-hadoop-cdh5.14.2.tar.gz -C /opt/modules/hadoop-2.6.0-cdh5.14.2/lib

在次检查支持解压

 可以观察到已经支持
 $ bin/hadoop checknative -a

配置hadoop (jobhistory可以查看所有配置信息)

mapred-site.xml 配置

    mapreduce.map.output.compress
    true



    mapreduce.map.output.compress.codec
    org.apache.hadoop.io.compress.SnappyCodec
    

测试

运行pi程序: $ bin/yarn jar share/hadoop/mapreduce/hadoop-mapreduce-examples-2.5.0-cdh5.3.6.jar pi 1 2

通过主机:19888观察该任务的configuration中压缩配置

通过hive配置压缩格式

shuffle阶段启用压缩

set hive.exec.compress.output=true;
set mapred.output.compress=true;
set mapred.output.compression.codec=org.apache.hadoop.io.compress.SnappyCodec;

​reduce输出的结果文件进行压缩

set mapreduce.output.fileoutputformat.compress=true;
set mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.SnappyCodec;

hive中的文件存储格式 格式

create table (
...
)
row format delimited fields terminated by ""
STORED AS file_format

文件格式如下

TEXTFILE

RCFILE

ORC

PARQUET

AVRO

INPUTFORMAT

常用的文件格式（默认是textfile ）

| ORC -- (Note: Available in Hive 0.11.0 and later)

| PARQUET --Parquet就是基于Dremel的数据模型和算法实现的。 这个比较常见

数据存储类型 按行存储

写的快

按列存储

读得快

可以跳过不符合条件的数据，只读取需要的数据，降低IO数据量。

压缩编码可以降低磁盘存储空间。由于同一列的数据类型是一样的，可以使用更高效的压缩编码进一步节约存储空间。

只读取需要的列，支持向量运算，能够获取更好的扫描性能。

验证存储格式及压缩

使用给定的日志文件（18.1MB）

使用不同的存储格式，存储相同的数据，判断文件大小

在MapReduce的shuffle阶段启用压缩(对中间数据进行压缩可以减少map和reduce task间的数据传输量。对于IO型作业，可以加快速度。)

set hive.exec.compress.intermediate=true;
set mapred.map.output.compression.codec=org.apache.hadoop.io.compress.SnappyCodec; 

对输出结果压缩

set hive.exec.compress.output=true;
set mapred.output.compression.codec=org.apache.hadoop.io.compress.SnappyCodec;

创建表file_text ，并加载数据

create table if not exists file_text(
t_time string,
t_url string,
t_uuid string,
t_refered_url string,
t_ip string,
t_user string,
t_city string
)
row format delimited fields terminated by "	"
stored  as  textfile;
load data local inpath "/home/hadoop/page_views.data" into table file_text;

对比默认格式和file_orc_snappy 数据大小比较

create table if not exists file_orc_snappy(
t_time string,
t_url string,
t_uuid string,
t_refered_url string,
t_ip string,
t_user string,
t_city string
)
row format delimited fields terminated by "	"
stored as  ORC
tblproperties("orc.compression"="Snappy");

insert into table file_orc_snappy select *  from file_text;
-- 不能通过load来加载，因为load本质是hdfs的put，这样不能压缩，必须要insert这样走MapReduce才能让压缩发挥作用

对比默认格式和parquet格式 数据大小比较

create table if not exists file_parquet(
t_time string,
t_url string,
t_uuid string,
t_refered_url string,
t_ip string,
t_user string,
t_city string
)
row format delimited fields terminated by "	"
stored as parquet;

insert into table file_parquet select * from file_text;

对比默认格式和parquet格式，snappy压缩 数据大小比较

create table if not exists file_parquet_snappy(
t_time string,
t_url string,
t_uuid string,
t_refered_url string,
t_ip string,
t_user string,
t_city string
)
row format delimited fields terminated by "	"
stored as parquet
tblproperties("parquet.compression"="Snappy");

insert into table file_parquet_snappy select * from file_text;

hive (mydb)> dfs -du -s -h /user/hive/warehouse/mydb.db/file_parquet_snappy;
hive (mydb)> dfs -du -s -h /user/hive/warehouse/mydb.db/file_parquet;       

hive中使用正则加载数据

通过正则匹配，加载复杂格式日志文件

1 正则

2 根据日志加载数据

​ 日志

"27.38.5.159" "-" "31/Aug/2015:00:04:53 +0800" "GET /course/view.php?id=27 HTTP/1.1" "200" "7877" - "http://www.ibf.com/user.php?act=mycourse&testsession=1637" "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/31.0.1650.63 Safari/537.36" "-" "learn.ibf.com"

​ 创建表

CREATE TABLE apachelog (

remote_addr string,

remote_user string,

time_local string,

request string,

status string,

body_bytes_set string,

request_body string,

http_referer string,

http_user_agent string,

http_x_forwarded_for string,

host string

)

ROW FORMAT SERDE "org.apache.hadoop.hive.serde2.RegexSerDe"

WITH SERDEPROPERTIES (

"input.regex" = "("[^ ]*") ("-|[^ ]*") ("[^]]*") ("[^]]*") ("[0-9]*") ("[0-9]*") (-|[^ ]*) ("[^ ]*") ("[^"]*") ("-|[^ ]*") ("[^ ]*")"

)

STORED AS TEXTFILE;

load data local inpath "/home/hadoop/moodle.ibf.access.log" into table apachelog;

Hive优化 大表拆分为小表 外部表+分区表 存储格式&数据压缩 SQL优化 并行执行

//Whether to execute jobs in parallel

set hive.exec.parallel=true;

//How many jobs at most can be executed in parallel

set hive.exec.parallel.thread.number=8;#可以调大,提高并行效率

mapreduce

Reduce数目

set mapreduce.job.reduces=1

JVM重用

mapreduce.job.jvm.numtasks=1 默认1个

推测执行

hive配置，默认为true

set hive.mapred.reduce.tasks.speculative.execution=true;

hadoop

mapreduce.map.speculative true

mapreduce.reduce.speculative true

设置输出文件合并

Size of merged files at the end of the job

将小文件合并避免降低hdfs存储大量小文件而降低性能

set hive.merge.size.per.task=256000000;

严格模式

set hive.mapred.mode=strict; nonstrict默认

​ 严格模式下，

​ 分区表，必须加分区字段过滤条件

​ 对order by, 必须使用limit

​ 限制笛卡尔积的查询（join 的时候不使用on，而使用where）

hive join

map join

如果关联查询两张表中有一张小表默认map join，将小表加入内存

hive.mapjoin.smalltable.filesize=25000000 默认大小

hive.auto.convert.join=true 默认开启

如果没有开启使用mapjoin，使用语句制定小表使用mapjoin

select /+ MAPJOIN(time_dim) / count(1) from

store_sales join time_dim on (ss_sold_time_sk = t_time_sk)

reduce join

对两张大表join

对关联的key进行分组

smb join

Sort-Merge-Bucket join

解决大表与大表join速度慢问题

通过分桶字段的的hash值对桶的个数取余进行分桶

set hive.enforce.bucketing=true;

create table 表名 (

字段

)

clustered by(分桶字段) into 分桶数量 buckets;

如

create table student(

id int,

age int,

name string

)

clustered by (id) into 4 bucket

row format delimited fields terminated by ",";

​

小文件处理

重建表，建表时减少reduce的数量

通过参数调节，设置map/reduce的数量

//每个Map最大输入大小(这个值决定了合并后文件的数量)

set mapred.max.split.size=256000000;

//一个节点上split的至少的大小(这个值决定了多个DataNode上的文件是否需要合并)

set mapred.min.split.size.per.node=100000000;

//一个交换机下split的至少的大小(这个值决定了多个交换机上的文件是否需要合并)

set mapred.min.split.size.per.rack=100000000;

//执行Map前进行小文件合并

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

//设置map端输出进行合并，默认为true

set hive.merge.mapfiles = true

//设置reduce端输出进行合并，默认为false

set hive.merge.mapredfiles = true

//设置合并文件的大小

set hive.merge.size.per.task = 256000000

//当输出文件平均大小小于设定值时，启动合并操作。这一设定只有当hive.merge.mapfiles或hive.merge.mapredfiles设定为true时，才会对相应的操作有效。

set hive.merge.smallfiles.avgsize=16000000

数据倾斜

本质原因：key的分布不均导致的

Map 端部分聚合，相当于Combiner

hive.map.aggr=true

有数据倾斜的时候进行负载均衡

hive.groupby.skewindata=true

当选项设定为 true，生成的查询计划会有两个 MR Job。第一个 MR Job 中，Map 的输出结果集合会随机分布到 Reduce 中，每个 Reduce 做部分聚合操作，并输出结果，这样处理的结果是相同的 Group By Key 有可能被分发到不同的 Reduce 中，从而达到负载均衡的目的；第二个 MR Job 再根据预处理的数据结果按照 Group By Key 分布到 Reduce 中（这个过程可以保证相同的 Group By Key 被分布到同一个 Reduce 中），最后完成最终的聚合操作。

hive案例：日志分析

名词
1) UV： count(distinct guid)
访问您网站的一台电脑客户端为一个访客。00:00-24:00内相同的客户端只被计算一次。
2) PV：Page View--- count(url)
即页面浏览量或点击量，用户每次刷新即被计算一次。
3) 登录人数：
登录网站访问的人数[会员]，endUserId有值的数量
4) 游客数：
没有登录访问的人数，endUserId为空的数量
5) 平均访问时长：

访客平均在网站停留的时间
trackTime  --> max - min

6) 二跳率： pv>1的访问量/总访问量

平均浏览2个页面及以上(pv>1)的用户数 /  用户总数(discont guid) 点击1次

二跳率的概念是当网站页面展开后，用户在页面上产生的首次点击被称为“二跳”，二跳的次数即为“二跳量”。二跳量与浏览量的比值称为页面的二跳率。

count(case when pv >=2 then guid else null end ) / discont (guid)

7) 独立IP：---count(distinct ip)
独立IP表示，拥有特定唯一IP地址的计算机访问您的网站的次数，因为这种统计方式比较容易实现，具有较高的真实性，所以成为大多数机构衡量网站流量的重要指标。比如你是ADSL拨号上网的，你拨一次号都自动分配一个ip，这样你进入了本站，那就算一个ip，当你断线了而没清理cookies，之后又拨 了一次号，又自动分配到一个ip，你再进来了本站，那么又统计到一个ip，但是UV（独立访客)没有变，因为2次都是你进入了本站。

日期	uv	pv	登录人数	游客人数	平均访问时间	二跳率	独立IP数

窗口函数分析函数 over语句

准备测试数据

hive (db_analogs)> create database ts;
hive (db_analogs)> use ts;
hive (ts)> create table testscore(gender string,satscore int, idnum int) row format delimited fields terminated by "	";
hive (ts)> load data local inpath "/opt/datas/TESTSCORES.csv" into table testscore;

OVER with standard aggregates: COUNT、SUM、MIN/MAX、 AVG

需求1：

按照性别分组，satscore分数排序（降序），最后一列显示所在分组中的最高分

Female  1000    37070397        1590
Female  970     60714297        1590
Female  910     30834797        1590
Male    1600    39196697        1600
Male    1360    44327297        1600
Male    1340    55983497        1600

答案sql:

hive (ts)>  select gender,satscore,idnum,max(satscore) over(partition by gender order by satscore desc) maxs  from testscore;

注 意：

partition by 是分组用的

分析函数

要求 topN

​ 按照性别分组，satscore排序（降序），最后一列显示在分组中的名次

需求1：

分数相同名次不同，名次后面根据行数增长

Female  1590    23573597        1
Female  1520    40177297        2
Female  1520    73461797        3
Female  1490    9589297 4
Female  1390    99108497        5
Female  1380    23048597        6  # 分数相同
Female  1380    81994397        7  # 分数相同

需求2：

分数相同名次相同，名次后面根据行数增长

Female  1590    23573597        1
Female  1520    40177297        2
Female  1520    73461797        2
Female  1490    9589297 4
Female  1390    99108497        5
Female  1380    23048597        6  #分数相同
Female  1380    81994397        6  # 分数相同

需求3：

分数相同名次相同，名次连续增长

Female  1590    23573597        1
Female  1520    40177297        2
Female  1520    73461797        2
Female  1490    9589297 3
Female  1390    99108497        4
Female  1380    23048597        5
Female  1380    81994397        5

SQL

sql1
hive (ts)> select gender,satscore,idnum,row_number() over(partition by gender order by satscore desc) maxs  from testscore;
-- ROW_NUMBER() 从1开始，按照顺序，生成分组内记录的序列

sql2
select gender,satscore,idnum,rank() over(partition by gender order by satscore desc) maxs  from testscore;
-- RANK() 生成数据项在分组中的排名，排名相等会在名次中留下空位 

sql3
select gender,satscore,idnum,dense_rank() over(partition by gender order by satscore desc) maxs  from testscore;
-- DENSE_RANK() 生成数据项在分组中的排名，排名相等会在名次中不会留下空位 

窗口函数

# 当有order by，而没有指定窗口子句时，窗口子句默认为RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW(从起点到当前行的范围)

# 当order by和窗口子句都没有时，窗口子句默认ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING（从起点到后面的终点）

UNBOUNDED PRECEDING

UNBOUNDED FOLLOWING

1 PRECEDING

1 FOLLOWING

CURRENT ROW

窗口对比

select gender,satscore,idnum,sum(satscore) over(partition by gender order by satscore desc RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) sums  from testscore;

select gender,satscore,idnum,sum(satscore) over(partition by gender order by satscore desc RANGE BETWEEN UNBOUNDED PRECEDING AND unbounded following) sums  from testscore;

select gender,satscore,idnum,sum(satscore) over(partition by gender order by satscore desc RANGE BETWEEN 1 PRECEDING AND CURRENT ROW) sums  from testscore;

当前行数据幅度+1后范围内

LAG

落后值（上n个值），在不指定落后个数的情况下，默认为落后一个值（数据从上向下显示，落后即当前值之前显示的值）

场景: 分析用户页面浏览顺序

sql

hive (ts)> select gender,satscore,idnum, lag(satscore) over(partition by gender order by satscore desc) as lastvalue from testscore;

要求

gender  satscore        idnum   lastvalue
Female  1590    23573597        NULL  # 此处为null，可以为其指定默认值
Female  1520    40177297        1590  # 显示当前satscore的上一条记录的值
Female  1520    73461797        1520  # 显示当前satscore的上一条记录的值
Female  1490    9589297 1520
Female  1390    99108497        1490

LEAD

与LAG相反（下n搁置），用法同理，前面的值（领先值），默认为领先一个值（数据从上向下显示，领先即当前值之后显示的值）

sql

hive (ts)> select gender,satscore,idnum, lead(satscore, 1, 0) over(partition by gender order by satscore desc) as nextvalue from testscore;

结果

gender  satscore        idnum   nextvalue
...
Female  1060    59149297        1060
Female  1060    46028397        1000
Female  1000    37070397        970
Female  970     60714297        910
Female  910     30834797        0