课程五：海量数据处理-Hive数据仓库

HADOOP大数据课程童小军2013年3月Hive数据仓库暴风公司数据仓库案例集群接收日志超过1.2TB/天3500+任务/日任务数据吞吐10TB+/天，离线小时数据分析数据挖掘和用户精分推荐系统用户按地域收入年龄性别收入层次划分广告系统数据系统的进化-一代数据系统的进化-二代数据系统的进化-三代暴风Hadoop集群架构流程Scribenginx+php使用的Hadoop生态圈Hadoop1.0.3基础计算框架

Hive离线数据分析95%任务

Pig离线数据分析HBase部分数据的存储Mahout数据挖掘暴风指数暴风指数暴风指数DataWarehousingatFacebook

WebServersScribeServersFilersHiveonHadoopClusterOracleRACFederatedMySQLWhatisHive

构建于hadoop的hdfs和mapred之上，用于管理和查询结构化/非结构化数据的数据仓库。使用HQL作为查询接口使用HDFS作为底层存储使用MapRed作为执行层WhatisHive

InstallationandConfiguration

安装和配置hadoop集群下载hive安装包解压至安装目录配置HADOOP_HOME和JAVA_HOME变量修改存储元数据的数据库（derby、mysql、oracle）

InstallationandConfiguration

hive的元数据存储hive默认使用内存数据库derby存储元数据，使用时不需要修改任何配置，缺点：hiveserver重启后所有的元数据都会丢失hive还执行mysql、oracle等任何支持JDBC连接方式的数据库来存储元数据，需要修改相应的配置项

InstallationandConfiguration

hive的元数据存储

DDLandDMLOperations

hive建表语法格式

DDLandDMLOperations

hive建表语法格式

external外部表，类似于mysql的csv引擎

partitionedby指定分区字段clusteredbysortedby可以对表和分区对某个列进行分桶操作，也可以利用sortedby对某个字段进行排序rowformat指定数据行中字段间的分隔符和数据行分隔符storedas指定数据文件格式：textfilesequencercfileinputformat(自定义的inputformat类)location指定数据文件存放的hdfs目录

DDLandDMLOperations

hive建表语句

CREATETABLEpage_view(viewTimeINT,useridBIGINT,page_urlSTRING,referrer_urlSTRING,ipSTRINGCOMMENT'IPAddressoftheUser')COMMENT'Thisisthepageviewtable'PARTITIONEDBY(dtSTRING,countrySTRING)CLUSTEREDBY(userid)SORTEDBY(viewTime)INTO32BUCKETSROWFORMATDELIMITEDFIELDSTERMINATEDBY'\001'COLLECTIONITEMSTERMINATEDBY'\002'MAPKEYSTERMINATEDBY'\003'STOREDASSEQUENCEFILE;

DDLandDMLOperations

删除表droptable[IFEXISTS]table_name删除内部表时会删除元数据和表数据文件删除外部表（external）时只删除元数据

DDLandDMLOperations

修改表增加分区

ALTERTABLEpage_viewADDPARTITION(dt='2008-08-08',country='us')location'/path/to/us/part080808'PARTITION(dt='2008-08-09',country='us')location'/path/to/us/part080809';修改表删除分区

ALTERTABLEpage_viewDROPPARTITION(dt='2008-08-08',country='us');修改表重命名表

ALTERTABLEtable_nameRENAMETOnew_table_name修改表修改字段

ALTERTABLEtest_changeCHANGEaa1STRINGAFTERb;

DDLandDMLOperations

加载数据

LOADDATAINPATH'/user/myname/kv2.txt'OVERWRITEINTOTABLEinvitesPARTITION(ds='2008-08-15');加载（本地、hdfs）文件到指定的表分区FROMsrc

INSERTOVERWRITETABLEdest1SELECTsrc.*WHEREsrc.key<100

INSERTOVERWRITETABLEdest2SELECTsrc.key,src.valueWHEREsrc.key>=100andsrc.key<200从指定表中选取数据插入到其他表中HQLusecase

select语法结构

SELECT[ALL|DISTINCT]select_expr,select_expr,...

FROMtable_reference

[WHEREwhere_condition]

[GROUPBYcol_list]

[CLUSTERBYcol_list|[DISTRIBUTEBYcol_list][SORTBYcol_list]]

[LIMITnumber]

HQLusecase

select案例

SELECTpv.pageid,u.ageFROMpage_viewpJOINuseruON(pv.userid=u.userid)JOINnewuserxon(u.age=x.age);

HiveExtension

User-definedFunctionUDF作用于单个数据行，输出一个数据，如字符处理函数UDAF作用于多个数据行，输出一个数据，如count，sum函数UDTF作用于单个数据行，输出多个数据行支持用户使用java自定义开发UDF函数Hivestreaming

支持用户在hiveQL语句中嵌入自定义的streaming处理脚本

使用UDFaddjar/tmp/helloUDF.jar;createtemporaryfunctionhelloworldas'com.hrj.hive.udf.HelloUDF';selecthelloworld(t.col1)fromtlimit10;droptemporaryfunctionhelloworld;HiveOptimization

使用分区Hive中的每个分区都对应hdfs上的一个目录，分区列也不是表中的一个实际的字段，而是一个或者多个伪列，在表的数据文件中实际上并不保存分区列的信息与数据。Partition关键字中排在前面的为主分区（只有一个），后面的为副分区静态分区：静态分区在加载数据和使用时都需要在sql语句中指定

案例：(stat_date='20120625',province='hunan')动态分区：使用动态分区需要设置hive.exec.dynamic.partition参数值为true，默认值为false，在默认情况下，hive会假设主分区时静态分区，副分区使用动态分区；如果想都使用动态分区，需要设置set

hive.exec.dynamic.partition.mode=nostrick，默认为strick

案例：(stat_date='20120625',province)HiveOptimization

排序优化

Order

by

实现全局排序，一个reduce实现，效率低Sort

by

实现部分有序，单个reduce输出的结果是有序的，效率高，通常和DISTRIBUTE

BY关键字一起使用（DISTRIBUTE

BY关键字

可以指定map

到

reduce端的分发key）CLUSTER

BY

col1

等价于DISTRIBUTE

BY

col1

SORT

BY

col1合并小文件文件数目过多，会给

HDFS

带来压力，并且会影响处理效率，可以通过合并

Map

和

Reduce

的结果文件来消除这样的影响hive.merge.mapfiles

=

true是否和并

Map

输出文件，默认为

Truehive.merge.mapredfiles

=

false是否合并

Reduce

输出文件，默认为

Falsehive.merge.size.per.task

=

256*1000*1000合并文件的大小HiveOptimization

groupby优化

Map端聚合，首先在map端进行初步聚合，最后在reduce端得出最终结果，相关参数：hive.map.aggr

=

true是否在

Map

端进行聚合，默认为

True

hive.groupby.mapaggr.checkinterval

=

100000在

Map

端进行聚合操作的条目数目数据倾斜聚合优化，设置参数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

中），最后完成最终的聚合操作。HiveOptimization

join优化

Join查找操作的基本原则：应该将条目少的表/子查询放在

Join

操作符的左边。原因是在

Join

操作的

Reduce

阶段，位于

Join

操作符左边的表的内容会被加载进内存，将条目少的表放在左边，可以有效减少发生内存溢出错误的几率。Join查找操作中如果存在多个join，且所有参与join的表中其参与join的key都相同，则会将所有的join合并到一个mapred程序中。案例：SELECT

a.val,

b.val,

c.val

FROM

a

JOIN

b

ON

(a.key

=

b.key1)

JOIN

c

ON

(c.key

=

b.key1)

在一个mapre程序中执行joinSELECT

a.val,

b.val,

c.val

FROM

a

JOIN

b

ON

(a.key

=

b.key1)

JOIN

c

ON

(c.key

=

b.key2)

在两个mapred程序中执行joinMap

join的关键在于join操作中的某个表的数据量很小，案例：SELECT

/*+

MAPJOIN(b)

*/

a.key,

a.value

FROM

a

join

b

on

a.key

=

b.key

HiveOptimization

join优化

Mapjoin

的限制是无法执行a

FULL/RIGHT

OUTER

JOIN

b，和map

join相关的hive参数：erval

hive.mapjoin.size.key

hive.mapjoin.cache.numrows由于join操作是在where操作之前执行，所以当你在执行join时，where条件并不能起到减少join数据的作用；案例：SELECT

a.val,

b.val

FROM

a

LEFT

OUTER

JOIN

b

ON

(a.key=b.key)

WHERE

a.ds='2009-07-07'

AND

b.ds='2009-07-07'最好修改为：SELECT

a.val,

b.val

FROM

a

LEFT

OUTER

JOIN

b

ON

(a.key=b.key

AND

b.ds='2009-07-07'

AND

a.ds='2009-07-07')在join操作的每一个mapred程序中，hive都会把出现在join语句中相对靠后的表的数据stream化，相对靠前的变的数据缓存在内存中。当然，也可以手动指定stream化的表：SELECT

/*+

STREAMTABLE(a)

*/

a.val,

b.val,

c.val

FROM

a

JOIN

b

ON

(a.key

=

b.key1)

JOIN

c

ON

(c.key

=

b.key1)HiveOptimization

实现（not）in通过left

outer

join进行查询（假设B表中包含另外的一个字段

key1

select

a.key

from

a

left

outer

join

b

on

a.key=b.key

where

b.key1

is

null通过left

semi

join

实现

inSELECT

a.key,

a.val

FROM

a

LEFT

SEMI

JOIN

b

on

(a.key

=

b.key)Left

semi

join

的限制：join条件中右边的表只能出现在join条件中。Hql使用自定义的mapred脚本注意事项：在使用自定义的mapred脚本时，关键字MAP

REDUCE

是语句SELECT

TRANSFORM

(

...

)的语法转换，并不意味着使用MAP关键字时会强制产生一个新的map过程，使用REDUCE关键字时会产生一个red过程。自定义的mapred脚本可以是hql语句完成更为复杂的功能，但是性能比hql语句差了一些，应该尽量避免使用，如有可能，使用UDTF函数来替换自定义的mapred脚本HiveOptimization

join优化

Mapjoin

的限制是无法执行a

FULL/RIGHT

OUTER

JOIN

b，和map

join相关的hive参数：erval

hive.mapjoin.size.key

hive.mapjoin.cache.numrows由于join操作是在where操作之前执行，所以当你在执行join时，where条件并不能起到减少join数据的作用；案例：SELECT

a.val,

b.val

FROM

a

LEFT

OUTER

JOIN

b

ON

(a.key=b.key)

WHERE

a.ds='2009-07-07'

AND

b.ds='2009-07-07'最好修改为：SELECT

a.val,

b.val

FROM

a

LEFT

OUTER

JOIN

b

ON

(a.key=b.key

AND

b.ds='2009-07-07'

AND

a.ds='2009-07-07')在join操作的每一个mapred程序中，hive都会把出现在join语句中相对靠后的表的数据stream化，相对靠前的变的数据缓存在内存中。当然，也可以手动指定stream化的表：SELECT

/*+

STREAMTABLE(a)

*/

a.val,

b.val,

c.val

FROM

a

JOIN

b

ON

(a.key

=

b.key1)

JOIN

c

ON

(c.key

=

b.key1)创建LZO压缩表CREATEEXTERNALTABLEfoo(columnAstring,columnBstring)PARTITIONEDBY(datestring)ROWFORMATDELIMITEDFIELDSTERMINATEDBY"\t"STOREDASINPUTFORMAT"com.hadoop.mapred.DeprecatedLzoTextInputFormat"OUTPUTFORMAT"org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat"LOCATION'/data/dw/uid/';HiveLzo文件加载流程压缩lzopuid.txt上传hadoopfs-putuid.txt.lzo/data/dw/uid/创建索引hadoopjar/path/to/your/hadoop-lzo.jarpression.lzo.DistributedLzoIndexer/data/dw/uid/uid.txt.lzo生成1000widshadoopjarcontrib/streaming/hadoop-streaming-1.0.3.jar-mapper'seq100000'-numReduceTasks0-input/user/hadoop/PiEstimator_TMP_3_141592654/in/-output/opt/ids/生成r.pl代码#!/usr/bin/perl-X$i=0;while($i<=10000){

$i++;

print`</dev/urandomtr-dc_A-Z-a-z-0-9|head-c10`."\n\r";}hadoopjarcontrib/streaming/hadoop-streaming-1.0.3.jar-mapper'cat'-reducer'seq10000'-numReduceTasks10-input/data/-output/data/test3/1.seq10>texthadoopfs-puttest/data/texthadoopjarcontrib/streaming/hadoop-streaming-1.0.3.jar-mapper'cat'-reducer'seq10000'-numReduceTasks10-input/data/text-output/data/test3/CREATEEXTERNALTABLEtest3(textstring)LOCATION'/data/test3/'Hivevssql

优化策略使用Partition减少扫描数量使用Map端Join配置Reduce数量xml,json提取适用脚本提取,而非使用函数使用INSERTINTOLOCALDIRECTORY‘/home/me/pv_age_sum.dir’,而非适用HiveServer。使用LZO压缩存储数据适用外部表,而非内部表press.output=false,true适用队列管理任务执行生成100亿车牌号hadoopjarhadoop-examples-1.0.3.jarpi10010000生成100个文件hadoopjarcontrib/streaming/hadoop-streaming-1.0.3.jar-numReduceTasks0-filerand.pl-mapperrand.pl-input/user/root/PiEstimator_TMP_3_141592654/in-output/opt/chepai/chepaicreateEXTERNALtablechepai(idbigint)ROWFORMATDELIMITEDFIELDSTERMINATEDBY'\t'collectionitemsterminatedby"\n"storedastextfilelocation'/opt/chepai/';Hive高可用措施健全监控体系使用负载均衡配置队列调度配置重启机制定时清理日志配置队列运行set=ETLselect*fromuid;-D=ETL运维和工具介绍-Cacti

Page

48Hadoop集群汇总监控GangliaHAProxy+Hive网络拓扑QueriesHAProxyHAProxyHiveHiveHiveHiveHadoopHAProxy+Hive高可用集群phpHiveAdmin界面

材料材料:.tw/cloud//downloads/sqoop/查询界面:任务调度/oozie///azkaban/screenshots.php数据交换/p/datax/wiki/DataX%E4%BA%A7%E5%93%81%E8%AF%B4%E6%98%8E//zhuyeqing/ComETL//cdh/3/sqoop/SqoopUserGuide.htmlhadoopwindows版本.tw/cloud/wiki/Hadoop4WinIntroductiontoPigPig是一个基于Hadoop的大规模数据分析平台，它提供的SQL-LIKE语言叫PigLatin，该语言的编译器会把类SQL的数据分析请求转换为一系列经过优化处理的MapReduce运算。Pig为复杂的海量数据并行计算提供了一个简单的操作和编程接口。PigSystemexecutionplanPigCompilerClusterparsedprogramParserusercross-joboptimizerPigLatinprogramMap-RMRCompilerjoinoutputfilterXf()YIntroductiontoPigPig的主要操作load操作：加载数据foreach操作：循环指定数据集中的所有数据filter操作：根据过滤条件筛选数据，通常和foreach配合使用group操作：分组操作，和sql中的groupby功能一致union操作：联合操作，和sql中的union功能一致join操作：join操作，和sql中的join功能一致store操作：将制定数据集存储至指定目录

IntroductiontoPigPig操作实例Users=LOAD‘users.in’AS(name,age);Fltrd=FILTERUsersbyage>=18andage<=25;

Pages=LOAD‘pages.in’AS(user,url);Jnd=JOINFltrdBYname,PagesBYuser;Grpd=GROUPJndbyurl;Smmd=FOREACHGrpdGENERATEgroup,COUNT(Jnd)ASclicks;Srtd=ORDERSmmdBYclicks;Top100=LIMITSrtd100;STORETop100INTO‘top100sites.out’;BI其他解决方案R+HadoopSqoop最佳实践第六次技术聚会分享HDFS到DBsqoopexport--connectjdbc:oracle:thin:@XXX:1521:XXX--usernameXXXX--passwordXXXXX--export-dir/user/flume/part-r-00000-m1--tableNS_SLLOG_CLIENT_SUM_OS--update-keyACTIVITYTIME,CLIENT_VER,OS--update-modeallowinsert--input-fields-terminated-by'\t'--outdirjobs--package-namecom.uusee.sqoop日志收集——flumeNG1）它是FlumeOG的重构版本，apache来开发管理。2）它变的很简单，Master、zookeeper、collector和WebUI没有了3）简化成了1、source（avro：很简单使用；exec：使用shell命令）2、sink（hdfs、file）3、channel(Memory、disk)FlumeNG——架构图FlumeNG——数据获取1）RPC1、在flume中，Avro客户端使用AVRORPC机制可以发送一个给定的文件

Avro

源：2、$bin/flume-ngavro-client-Hlocalhost-p

41414

-F/usr/logs/log.103、上面的命令将发送的/usr/logs/log.10的内容到

flume源监听端2）Executingcommands1、还有一个exec执行一个给定的命令获得输出的源。一个单一的输出，即“line”。回车（'\R'）或换行符（'\N