大数据学习笔记之Hive（四）：Hive关于表的相关操作、Sqoop

文章目录

• Hive创建表的方式
• Hive表导入数据方式
• Hive表导出数据方式
• Hive数据清洗之思路
• Hive自动化处理日志
• Hive中的几种排序
• 业务案例梳理
• Sqoop
• 一、安装：
• 二、配置：
• 1、开启Zookeeper
• 2、开启集群服务
• 3、配置文件：
• 4、拷贝jdbc驱动到sqoop的lib目录下
• 5、启动sqoop
• 6、测试Sqoop是否能够连接成功
• 三、 案例
• RDBMS --> HDFS
• 全部导入
• 查询导入
• 导入指定列
• 使用sqoop关键字筛选查询导入数据
• RDBMS --> Hive
• Hive/HDFS --> RDBMS

Hive创建表的方式

1、使用create命令创建一个新表
例如：create table if not exists db_web_data.track_log(字段)
partitioned by (date string,hour string)
row format delimited fields terminated by ‘\t’;
2、把一张表的某些字段抽取出来，创建成一张新表
例如：create table backup_track_log as select * from db_web_data.track_log;
尖叫提示：会复制属性以及属性值到新的表中
3、复制表结构
例如：create table like_track_log like db_web_data.track_log;
尖叫提示：不会复制属性值，只会复制表结构。

Hive表导入数据方式

1、本地导入
load data local inpath ‘local_path/file’ into table 表名称 ;
2、HDFS导入
load data inpath ‘hdfs_path/file’ into table 表名称 ;
3、覆盖导入
load data local inpath ‘path/file’ overwrite into table 表名称 ;
load data inpath ‘path/file’ overwrite into table 表名称 ;
4、查询导入
create table track_log_bak as select * from db_web_data.track_log;
5、insert导入–常用
** 追加-append-默认方式
insert into table 表名 select * from track_log;
** 覆盖-overwrite-显示指定-使用频率高
insert overwrite table 表名 select * from track_log;

Hive表导出数据方式

1、本地导出
例如：insert overwrite local directory “/home/admin/Desktop/1/2” row format delimited fields terminated by ‘\t’ select * from db_hive_demo.emp ;
insert overwrite local directory "/home/admin/Desktop/1/2"指定一个本地的目录， select * from db_hive_demo.emp ;查询，将查询出来的结果全部放到对应的目录下，中间的不用管。
尖叫提示：会递归创建目录，但是注意home目录其他用户是没有可写权限的，只有root用户有。
2、HDFS导出
例如：insert overwrite diretory “path/” select * from staff;
3、Bash shell覆盖追加导出
例如：$ bin/hive -e “select * from staff;” > /home/z/backup.log
4、Sqoop

Hive数据清洗之思路

例如在20150828的日志里面，有一些错乱的数据，比如查询出来的数据没有url，或者省份id和期望的值不一样，有的省份id是汉字是不合法的

需求：实现按照省份和日期统计PV和UV
SELECT date, provinceId, count(url) pv, count(distinct guid) uv from track_log group by date, provinceId;
见代码
![在这里插入

省份中有汉字，是不合法的

上图存在url为空的情况，这个时候通过split("\t")的时候，可能出现只有两个tab键，这个时候就少了一个字段相当于，选择过滤掉这条数据。

注意：下面的代码中可能有大量的错误，主要看逻辑，主要看LogCleanMapper，看是如何数据清洗的。

public cLass LogCleanMapper extends Mapper<Longwritable, Text, Longwritable, Text> {
	@OVerride
	protected void map(Longwritable key, Text value, Contextcontext) throws IOException, InterruptedException{

		//上图存在url为空的情况，这个时候通过split("\t")的时候，可能出现只有两个tab键，这个时候就少了一个字段相当于，选择过滤掉这条数据。
		//所以判断字段，如果是空的或者是“”的，将不将这条数据写入MapReduce
		
		String line value tostring()
		//id guid provinceId
		String[] splits Line split(\t")
		if(splits, length 32){
			context. getCounter("Web Data ETL","Data Length is too short"). increment(lL);:
		return:
		}
		String url splits[l]:
		if(StringUtils, isBLank(urL)){
			context. getCounter("Web Data ETL",url is blank"). inc rement(IL);
			return;
		}
		String provinceId spLits[20]:
		int provinceIdInt Integer MAX VALUE;
		try {
		//这里对省份还进行了一次处理，如果省份是汉字的话经过pareInt函数的时候会报错
		//通过这种方法将省份字段格式错误的过滤掉
		//不将这行写入MapReduce
			provinceIdInt Integer parseInt(provinceId):
		}catch (Exception e){
			context. getCounter("Web Data ETL","provinceId is UNKNOWN"). inc rement(1L);
			return:
		}

		if(StringUtils. isBLank(provinceId)){
			context. getCounter("Web Data ETL","provinck ).increment(1L);
			return
		}
		context.write(key,value);
	}
}

public cLass LogCleanReduce extends Reducer<LongWritable, Text, Text, NulLwritable> {
oVerride
	protected void reduce(Longwritable key, Iterable<Text> values, Context context)throws IOException, Ir...){
		for(Text value values)t
		context. write(value, NulLWritable get());
	}
}

public cLass LogCLeanMapReduce extends Configured impLements Tool
public static void main (String[] args )throws Exceptiont
	if(args Length < 0){
		args new String[]t " hdfs: //hadoop-seniorol itguigu com/input/words. txt","hdfs: //hadoop-sen
		args new String[]["/input/words. txt","/output/"1:
	}
	pubLic int run (String[] args) throws Exception {
		Configuration conf new Configuration ()
		Job job Job getInstance(conf)
		job. setJarBy CLass(LogcleanMapReduce. cLass);
		Path inPath new Path(args[0]);
		FileInputFormat. addInputPath(job, inPath):
		job. setMapperCLass(LogCleanMapper. cLass);
		job. setMapOutputKey CLass( Longwritable. cLass);
		job. setMapoutputValueClass(Text. cLass);
		job. setReducerCLass(LogClean Reduce class):
		job. setoutputKey Class(Text. cLass);
		job. setoutputValueCLass(NulLWritable cLass);
		Path outPath new Path(args[1]):
		FiLeOutputFormat. setOutputPath(job, outPath):
		booLean isSuccess job. waitForCompletion(true);
		return insuccess ? 0: 1
		}
}

Hive自动化处理日志

需求：网站产生的日志会按照某个日期分类的文件夹存储，定期分析该日志，产出结果

假如在如下的目录中存放着日志

/opt/modules/weblog
	-20170724(文件夹)
		-2017072418（日志文件）
		-2017072419
	-20170725
		-2017072518（日志文件）
		-2017072619

见脚本

auto.sh

#!/bin/bash
# 如何执行系统变量？ . /etc/profile
#执行系统环境变量脚本，初始化一些变量信息
. /etc/profile

#定义Hive目录
HIVE_DIR=/opt/modules/cdh/hive-0.13.1-cdh5.3.6

#定义日志的存储路径
WEB_LOG=/opt/modules/weblog

#昨天的日期，用于访问目录
# %Y%m%d用于格式化
YESTERDAY=$(date --date="1 day ago" +%Y%m%d)

#遍历目录
for i in `ls $WEB_LOG/$YESTERDAY`
do
# 切片 2015082818 
	DATE=${i:0:8}
# 10150828 从第零位开始，截取八位
	HOUR=${i:8:2}
# 18 从第八位开始，截取两位
# 导入数据
//这里的"load data local inpath "$WEB_LOG/$YESTERDAY/$i" 。。。。“
//其实是字符串拼接，拼接命令，”xxx“变量”xxx“，相当于java中的字符串拼接
//可是hive -e 后面的内容也需要引号，总不能hive -e ””xxx“ 变量 ”xxx““  把
//所以hive中有一个hiveconf，能够把变量传入到文件中
	#$HIVE_DIR/bin/hive -e "load data local inpath "$WEB_LOG/$YESTERDAY/$i" 
	#into table db_web_log.track_log partition (date='$DATE', hour='$HOUR')"
	$HIVE_DIR/bin/hive \
	--hiveconf LOADFILE_NEW=$WEB_LOG/$YESTERDAY/$i \
	--hiveconf DATE_NEW=$DATE \
	--hiveconf HOUR_NEW=$HOUR \
	-f $HIVE_DIR/hql/auto.hql
done

auto.hql

load data local inpath "${hiveconf:LOADFILE_NEW}" into table db_web_log.track_log partition(data='${hiveconf:DATE_NEW}'),hour='${hiveconf:HOUR_NEW}');

注意上面脚本中的切片中的第一个参数必须是变量不能直接写进去或者单双引号包住的形式。



数据的分析，数据的清洗，数据的转移等都可以通过这种方式配合定时任务达到想要的。

Hive中的几种排序

order by
全局排序，就一个Reduce
sort by
相当于对每一个Reduce内部的数据进行排序，不是全局排序。
distribute by
类似于MRpartition， 进行分区，一般要结合sort by使用。分成的区就是Reduce
cluster by
当distribute和sort字段相同时，就是cluster by，意思就是distribute和sort后面跟的是一个字段，可以直接写成cluster by 这个字段
案例见：HQL案例.txt

insert overwrite local directory "/home/admin/result/order" row format delimited fields terminated by "/t"
select * from emp order by empno;

//distribute英文的意思分开、散步、把。。。分类，按照部门编号把信息归到一个类中，有点类似部门相同的人分到一个分区，会产生若干文件，每个文件中都是相同部门的人。

insert overwrite local directory "/home/admin/result/sort" row format delimited fields terminated by "/t"
select * from emp distribute by deptno sort by empno

怎么查在任务执行的过程中涉及到的参数


如上图中的参数，可以通过hive客户端临时设置

如果直接运行命令set mapreduce.job.reduces; 会显示当前的设置，运行set mapreduce.job.reduces = 3;是设置将要改变的值。
设只为三个之前，distribute by 和前面的 order by是没有区别的，如下图所示

设置为mapreduce.job.reduces = 3;之后再次运行上面的distribute语句，变成了三个，为什么是三个？因为上面的语句根据deptno分成了三个。

业务案例梳理

需求：执行周期性任务，每天的晚上6点，执行自动化脚本，
加载昨天的日志文件到HDFS，
同时分析网站的多维数据（PV,UV按照省份和小时数进行分类查询）
最后将查询的结果，存储在一张临时表中（表字段：date，hour，provinceId，pv，uv）存储在HIVE中，并且
将该临时表中的所有数据，存储到MySQL中，以供第二天后台开发人员的调用，展示。
1、定时加载本地数据到HDFS，涉及到：auto.sh，crontab
2、清洗数据，打包jar，定时执行，
/user/hive/warehouse/db_web_data.db/track_log/date=20150828/hour=18
part-000001
/user/hive/warehouse/db_web_data.db/track_log/date=20150828/hour=19
part-000001
3、建表track_log，也不需要建立现成的分区，临时指定清洗好的数据作为仓库源
alter table track_log add partition(date=‘20150828’,hour=‘18’) location
“/user/hive/warehouse/db_web_data.db/track_log/date=20150828/hour=18”;
alter table track_log add partition(date=‘20150828’,hour=‘18’) location
“/user/hive/warehouse/db_web_data.db/track_log/date=20150828/hour=19”;
4、开始分析想要的数据，将结果存储在Hive的临时表中
创建临时表：
create table if not exists temp_track_log(date string, hour string, provinceId string, pv string, uv string)
row format delimited fields terminated by ‘\t’;
向临时表中插入数据：
insert overwrite table temp_track_log select date, hour, provinceId, count(url) pv, count(distinct guid) uv
from track_log where date=‘20150828’ group by date, hour, provinceId;
5、使用自定义的JAR，导入本地导出的文件到MYsql或者使用Sqoop。

Sqoop

sqoop相当于一个把数据倒来倒去工具箱，相当于FileZilla（mac和linux文件互传的工具），
SQL-TO-HADOOP
HDFS-----mysql（传文件）
mysql-----Hive（传文件）

一、安装：

直接解压即可

将conf目录下的sqoop-env-template.sh改名字为sqoop-env.sh

删除conf目录下所有cmd结尾的文件

二、配置：

1、开启Zookeeper

修改原来的脚本，新加开启zookeeper的脚本 start-cluster.sh

jps检查是否开启

2、开启集群服务

上面的脚本中也包括开启集群服务

3、配置文件：

** sqoop-env.sh

#export HADOOP_COMMON_HOME=
#hadoop的绝对路径
export HADOOP_COMMON_HOME=/opt/modules/cdh/hadoop-2.5.0-cdh5.3.6/

#Set path to where hadoop-*-core.jar is available
#export HADOOP_MAPRED_HOME=
export HADOOP_MAPRED_HOME=/opt/modules/cdh/hadoop-2.5.0-cdh5.3.6/

#set the path to where bin/hbase is available
#还没学先不管
#export HBASE_HOME=

#Set the path to where bin/hive is available
#export HIVE_HOME=
#HIVE路径
export HIVE_HOME=/opt/modules/cdh/hive-0.13.1-cdh5.3.6/

#Set the path for where zookeper config dir is
#export ZOOCFGDIR=
export ZOOCFGDIR=/opt/modules/cdh/zookeeper-3.4.5-cdh5.3.6/
export ZOOKEEPER_HOME=/opt/modules/cdh/zookeeper-3.4.5-cdh5.3.6/

一会可能会有一个警告，因为hbase没有配置，可以先不用管

4、拷贝jdbc驱动到sqoop的lib目录下

cp -a mysql-connector-java-5.1.27-bin.jar /opt/modules/cdh/sqoop-1.4.5-cdh5.3.6/lib/

5、启动sqoop

$ bin/sqoop help

查看帮助
貌似就是通过这个命令启动

6、测试Sqoop是否能够连接成功

$ bin/sqoop list-databases --connect jdbc:mysql://hadoop-senior01.itguigu.com:3306/metastore 
--username root \
--password 123456

展示出了mysql中的所有数据库列表，成功了

三、 案例

tips：在hadoop中规定，mysql数据库往hdfs中导叫导入，从hdfs往外面导叫导出

使用sqoop将mysql中的数据导入到HDFS

Step1、确定Mysql服务的正常开启
Step2、在Mysql中创建一张表

mysql> create database company;
mysql> create table staff(
id int(4) primary key not null auto_increment,
name varchar(255) not null,
sex varchar(255) not null);
mysql> insert into staff(name, sex) values(‘Thomas’, ‘Male’);
Step3、操作数据

RDBMS --> HDFS

使用Sqoop导入数据到HDFS

全部导入

//记得不要忘了换行的时候在后面加”\“符号
$ bin/sqoop import \
	--connect jdbc:mysql://hadoop-senior01.itguigu.com:3306/company \
	--username root \
	--password 123456 \
	--table staff \
	--target-dir /user/company \
	--delete-target-dir \
	--num-mappers 1 \
	--fields-terminated-by "\t"

–delete-target-dir \ 如果目录存在的话就删除掉它，在导之前提前删除它；
–num-mappers 1 \ 用几个map去跑这件事情
–fields-terminated-by “\t” 用什么分割

记得执行之前在hdfs中创建对应的目录

执行完之后查看结果：


导出的时候记得指定分隔符为\t，也方便将数据重新指向到hive仓库中

查询导入

//运行时记得每行后面加\
$ bin/sqoop import 
	 --connect jdbc:mysql://hadoop-senior01.itguigu.com:3306/company 
	 --username root 
	 --password 123456 
	 --target-dir /user/company 
	 --delete-target-dir 
	 --num-mappers 1 
	 --fields-terminated-by "\t" 
	 --query 'select name,sex from staff where id >= 2 and $CONDITIONS;'

意思是将查询出来的结果导入到hdfs中
末尾要加上and $CONDITIONS ，如果不加的话会报如下的错误

导入指定列

//运行时记得每行后面加\
$ bin/sqoop import 
	 --connect jdbc:mysql://hadoop-senior01.itguigu.com:3306/company 
	 --username root 
	 --password 123456 
	 --target-dir /user/company 
	 --delete-target-dir 
	 --num-mappers 1 
	 --fields-terminated-by "\t"
	 --columns id, sex
	 --table staff

使用sqoop关键字筛选查询导入数据

//运行时记得每行后面加\
$ bin/sqoop import 
	 --connect jdbc:mysql://hadoop-senior01.itguigu.com:3306/company 
	 --username root 
	 --password 123456 
	 --target-dir /user/company 
	 --delete-target-dir 
	 --num-mappers 1 
	 --fields-terminated-by "\t"
	 --table staff
	 --where "id=3"

RDBMS --> Hive

1、在Hive中创建表（不需要提前创建表，会自动创建，所以这步可以忽略）

hive (company)> create table staff_hive(id int, name string, sex string) row format delimited fields terminated by '\t';

hive中创建的表的结构要和弄出来的列一样，不是和mysql的列一样，加入，最后只需要id name，mysql中有十个字段，hive创建的表结构就是id name

2、向Hive中导入数据

//运行时记得每行后面加\
$ bin/sqoop import 
	--connect jdbc:mysql://hadoop-senior01.itguigu.com:3306/company 
	--username root 
	--password 123456 
	--table staff 
	--num-mappers 1 
	--hive-import 
	--fields-terminated-by "\t" 
	--hive-overwrite 
	--delete-target-dir
	--hive-table company.staff_hive

hive-import 指定向hive中导入
hive-overwrite覆盖
delete-target-dir如果目录已经存在还往里面导入是导入不成功的会报错（这一行是可选的）
hive-table company.staff_hive
hive-table company.staff_hive指定hive中的哪个库

可以看到也是通过MapReduce来跑这件事情

其实就是先将mysql移入到hdfs总，在将数据移入到对应的仓库中

Hive/HDFS --> RDBMS

1、在Mysql中创建一张表

$ bin/sqoop export 
--connect jdbc:mysql://hadoop-senior01.itguigu.com:3306/company 
--username root 
--password 123456
--table staff_mysql
--num-mappers 1 
--export-dir /user/hive/warehouse/company.db/staff_hive
--input-fields-terminated-by "\t" 

/user/hive/warehouse/company.db/staff_hive指的是hdfs中的目录

使用opt文件打包sqoop命令，然后执行
Step1、创建一个.opt文件
Step2、编写sqoop脚本

export
	--connect
	jdbc:mysql://hadoop-senior01.itguigu.com:3306/company
	--username
	root
	--password
	123456
	--table
	staff_mysql
	--num-mappers 1 
	--export-dir
	/user/hive/warehouse/company.db/staff_hive
	--input-fields-terminated-by
	"\t"

Step3、执行该脚本

$ bin/sqoop --options-file opt/job_hffs2rdbms.opt