課程五：海量數據處理-Hive數據倉庫

HADOOP大數據課程童小軍2013年3月Hive數據倉庫暴風公司數據倉庫案例集群接收日志超過1.2TB/天3500+任務/日任務數據吞吐10TB+/天，離線小時數據分析數據挖掘和用戶精分推薦系統(tǒng)用戶按地域收入年齡性別收入層次劃分廣告系統(tǒng)數據系統(tǒng)的進化-一代數據系統(tǒng)的進化-二代數據系統(tǒng)的進化-三代暴風Hadoop集群架構流程Scribenginx+php使用的Hadoop生態(tài)圈Hadoop1.0.3基礎計算框架

Hive離線數據分析95%任務

Pig離線數據分析HBase部分數據的存儲Mahout數據挖掘暴風指數暴風指數暴風指數DataWarehousingatFacebook

WebServersScribeServersFilersHiveonHadoopClusterOracleRACFederatedMySQLWhatisHive

構建于hadoop的hdfs和mapred之上，用于管理和查詢結構化/非結構化數據的數據倉庫。使用HQL作為查詢接口使用HDFS作為底層存儲使用MapRed作為執(zhí)行層WhatisHive

InstallationandConfiguration

安裝和配置hadoop集群下載hive安裝包解壓至安裝目錄配置HADOOP_HOME和JAVA_HOME變量修改存儲元數據的數據庫（derby、mysql、oracle）

InstallationandConfiguration

hive的元數據存儲hive默認使用內存數據庫derby存儲元數據，使用時不需要修改任何配置，缺點：hiveserver重啟后所有的元數據都會丟失hive還執(zhí)行mysql、oracle等任何支持JDBC連接方式的數據庫來存儲元數據，需要修改相應的配置項

InstallationandConfiguration

hive的元數據存儲

DDLandDMLOperations

hive建表語法格式

DDLandDMLOperations

hive建表語法格式

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

partitionedby指定分區(qū)字段clusteredbysortedby可以對表和分區(qū)對某個列進行分桶操作，也可以利用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

修改表增加分區(qū)

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';修改表刪除分區(qū)

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）文件到指定的表分區(qū)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自定義開發(fā)UDF函數Hivestreaming

支持用戶在hiveQL語句中嵌入自定義的streaming處理腳本

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

使用分區(qū)Hive中的每個分區(qū)都對應hdfs上的一個目錄，分區(qū)列也不是表中的一個實際的字段，而是一個或者多個偽列，在表的數據文件中實際上并不保存分區(qū)列的信息與數據。Partition關鍵字中排在前面的為主分區(qū)（只有一個），后面的為副分區(qū)靜態(tài)分區(qū)：靜態(tài)分區(qū)在加載數據和使用時都需要在sql語句中指定

案例：(stat_date='20120625',province='hunan')動態(tài)分區(qū)：使用動態(tài)分區(qū)需要設置hive.exec.dynamic.partition參數值為true，默認值為false，在默認情況下，hive會假設主分區(qū)時靜態(tài)分區(qū)，副分區(qū)使用動態(tài)分區(qū)；如果想都使用動態(tài)分區(qū)，需要設置set

hive.exec.dynamic.partition.mode=nostrick，默認為strick

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

排序優(yōu)化

Order

by

實現(xiàn)全局排序，一個reduce實現(xiàn)，效率低Sort

by

實現(xiàn)部分有序，單個reduce輸出的結果是有序的，效率高，通常和DISTRIBUTE

BY關鍵字一起使用（DISTRIBUTE

BY關鍵字

可以指定map

到

reduce端的分發(fā)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優(yōu)化

Map端聚合，首先在map端進行初步聚合，最后在reduce端得出最終結果，相關參數：hive.map.aggr

=

true是否在

Map

端進行聚合，默認為

True

hive.groupby.mapaggr.checkinterval

=

100000在

Map

端進行聚合操作的條目數目數據傾斜聚合優(yōu)化，設置參數hive.groupby.skewindata

=

true，當選項設定為

true，生成的查詢計劃會有兩個

MR

Job。第一個

MR

Job

中，Map

的輸出結果集合會隨機分布到

Reduce

中，每個

Reduce

做部分聚合操作，并輸出結果，這樣處理的結果是相同的

Group

By

Key

有可能被分發(fā)到不同的

Reduce

中，從而達到負載均衡的目的；第二個

MR

Job

再根據預處理的數據結果按照

Group

By

Key

分布到

Reduce

中（這個過程可以保證相同的

Group

By

Key

被分布到同一個

Reduce

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

join優(yōu)化

Join查找操作的基本原則：應該將條目少的表/子查詢放在

Join

操作符的左邊。原因是在

Join

操作的

Reduce

階段，位于

Join

操作符左邊的表的內容會被加載進內存，將條目少的表放在左邊，可以有效減少發(fā)生內存溢出錯誤的幾率。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程序中執(zhí)行joinSELECT

a.val,

b.val,

c.val

FROM

a

JOIN

b

ON

(a.key

=

b.key1)

JOIN

c

ON

(c.key

=

b.key2)

在兩個mapred程序中執(zhí)行joinMap

join的關鍵在于join操作中的某個表的數據量很小，案例：SELECT

/*+

MAPJOIN(b)

*/

a.key,

a.value

FROM

a

join

b

on

a.key

=

b.key

HiveOptimization

join優(yōu)化

Mapjoin

的限制是無法執(zhí)行a

FULL/RIGHT

OUTER

JOIN

b，和map

join相關的hive參數：erval

hive.mapjoin.size.key

hive.mapjoin.cache.numrows由于join操作是在where操作之前執(zhí)行，所以當你在執(zhí)行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都會把出現(xiàn)在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

實現(xiàn)（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

實現(xiàn)

inSELECT

a.key,

a.val

FROM

a

LEFT

SEMI

JOIN

b

on

(a.key

=

b.key)Left

semi

join

的限制：join條件中右邊的表只能出現(xiàn)在join條件中。Hql使用自定義的mapred腳本注意事項：在使用自定義的mapred腳本時，關鍵字MAP

REDUCE

是語句SELECT

TRANSFORM

(

...

)的語法轉換，并不意味著使用MAP關鍵字時會強制產生一個新的map過程，使用REDUCE關鍵字時會產生一個red過程。自定義的mapred腳本可以是hql語句完成更為復雜的功能，但是性能比hql語句差了一些，應該盡量避免使用，如有可能，使用UDTF函數來替換自定義的mapred腳本HiveOptimization

join優(yōu)化

Mapjoin

的限制是無法執(zhí)行a

FULL/RIGHT

OUTER

JOIN

b，和map

join相關的hive參數：erval

hive.mapjoin.size.key

hive.mapjoin.cache.numrows由于join操作是在where操作之前執(zhí)行，所以當你在執(zhí)行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都會把出現(xiàn)在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)創(chuàng)建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/創(chuàng)建索引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

優(yōu)化策略使用Partition減少掃描數量使用Map端Join配置Reduce數量xml,json提取適用腳本提取,而非使用函數使用INSERTINTOLOCALDIRECTORY‘/home/me/pv_age_sum.dir’,而非適用HiveServer。使用LZO壓縮存儲數據適用外部表,而非內部表press.output=false,true適用隊列管理任務執(zhí)行生成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高可用措施健全監(jiān)控體系使用負載均衡配置隊列調度配置重啟機制定時清理日志配置隊列運行set=ETLselect*fromuid;-D=ETL運維和工具介紹-Cacti

Page

48Hadoop集群匯總監(jiān)控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的大規(guī)模數據分析平臺，它提供的SQL-LIKE語言叫PigLatin，該語言的編譯器會把類SQL的數據分析請求轉換為一系列經過優(yōu)化處理的MapReduce運算。Pig為復雜的海量數據并行計算提供了一個簡單的操作和編程接口。PigSystemexecutionplanPigCompilerClusterparsedprogramParserusercross-joboptimizerPigLatinprogramMap-RMRCompilerjoinoutputfilterXf()YIntroductiontoPigPig的主要操作load操作：加載數據foreach操作：循環(huán)指定數據集中的所有數據filter操作：根據過濾條件篩選數據，通常和foreach配合使用group操作：分組操作，和sql中的groupby功能一致union操作：聯(lián)合操作，和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來開發(fā)管理。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機制可以發(fā)送一個給定的文件

Avro

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

41414

-F/usr/logs/log.103、上面的命令將發(fā)送的/usr/logs/log.10的內容到

flume源監(jiān)聽端2）Executingcommands1、還有一個exec執(zhí)行一個給定的命令獲得輸出的源。一個單一的輸出，即“l(fā)ine”?；剀嚕?\R'）或換行符（'\N