基于Hadoop生態(tài)系統(tǒng)的大數(shù)據(jù)解決方案TP05

第五章Hadoop綜合應(yīng)用預習檢查請說出TableMapper與Mapper的區(qū)別請說出TableReducer與Reducer的區(qū)別如何解決客戶端和任務(wù)的第三方依賴庫問題？在Job運行期間，共享全局變量有哪些方法？集中測試2/41回顧與作業(yè)點評簡述HMaster、HRegionServer、HRegion間的關(guān)系描述HBase概念視圖的結(jié)構(gòu)HBaseShell常用操作有哪些？HBase數(shù)據(jù)存儲管理API有哪些？點評作業(yè)的提交情況和共性問題回顧作業(yè)點評3/41本章任務(wù)使用HDFS+MapReduce+HBase實現(xiàn)音樂排行榜4/41本章目標理解HBase和MapReduce的集成使用掌握快速導入數(shù)據(jù)到HBase掌握MapReduce接收HBase數(shù)據(jù)輸入掌握MapReduce輸出到HBase5/41MapReduce與HBase集成策略使用MapReduce計算使用HBase存儲、查詢數(shù)據(jù)優(yōu)勢對HBase中的數(shù)據(jù)進行非實時性的統(tǒng)計分析對HBase中的數(shù)據(jù)進行分布式計算在多個MapReduce間使用HBase作為存儲媒介方法使用HBaseMapReduceAPI6/41HBaseMapReduceAPI簡介位于HBase提供的“hbase-server-1.0.2.jar”繼承自HadoopMapReduceAPI需在Hadoop集群中配置HBase的環(huán)境復制hbase-site.xml到$HADOOP_HOME/etc/hadoop編輯$HADOOP_HOME/etc/hadoop/hadoop-env.shexportHADOOP_CLASSPATH=$HADOOP_CLASSPATH:~/hbase-1.0.2/lib/*HBaseMapReduceAPIHadoopMapReduceAPIorg.apache.hadoop.hbase.mapreduce.TableMapperorg.apache.hadoop.mapreduce.Mapperorg.apache.hadoop.hbase.mapreduce.TableReducerorg.apache.hadoop.mapreduce.Reducerorg.apache.hadoop.hbase.mapreduce.TableInputFormatorg.apache.hadoop.mapreduce.InputFormatorg.apache.hadoop.hbase.mapreduce.TableOutputFormatorg.apache.hadoop.mapreduce.OutputFormat演示HBase在Hadoop中的環(huán)境配置7/41HBaseMapReduce應(yīng)用示例“hbase-server-1.0.2.jar”提供的應(yīng)用rowcounterimporttsvcompletebulkload示例hadoopjar~/hbase-1.0.2/lib/hbase-server-1.0.2.jarrowcountermusic示例hadoopjar~/hbase-1.0.2/lib/hbase-server-1.0.2.jarimporttsv–Dimporttsv.bulk.output=tmp–Dimporttsv.columns=HBASE_ROW_KEY,info:name,info:singer,info:gender,info:ryghme,info:terminalmusic/input2/music2示例hadoopjar~/hbase-1.0.2/lib/hbase-server-1.0.2.jarcompletebulkloadtmpmusicHBase表名指定生成的HFile目錄,“/user/hduser/tmp”“/user/hduser/tmp”和importtsv結(jié)合使用實現(xiàn)批量數(shù)據(jù)導入HDFS目錄，表示數(shù)據(jù)來源演示行數(shù)統(tǒng)計和批量導入8/41學員操作——HBase批量導入需求說明獲取素材music1.txt、music2.txt、music3.txt使用批量導入工具導入音樂播放記錄，其中表名：music行鍵格式：ID_NAME_DATE列族：info列鍵：info:name，info:singer，info:gender，info:ryghme，info:terminal實現(xiàn)思路將hbase-site.xml復制到Hadoop配置目錄將HBase類庫中的jar文件添加到$HADOOP_CLASSPATH環(huán)境變量中運行hbase-server-1.0.2.jar中的importtsv和completebulkload作業(yè)導入數(shù)據(jù)完成時間：20分鐘練習9/41常見問題及解決辦法代碼規(guī)范問題調(diào)試技巧共性問題集中講解共性問題集中講解10/41編寫HBaseMapReduce依賴的jar文件HBase

jar文件MapReducejar文件MapReduce編程模型MapperReducerJob涉及到的類TableMapperTableInputFormatTableOutputFormatTableReducerTableMapReduceUtil11/41類定義為什么TableMapper不提供輸入鍵/值類型？TableInputFormat繼承InputFormat<ImmutableBytesWritable,Result>讀取HBase行數(shù)據(jù)并封裝為<ImmutableBytesWritable,Result>并傳遞給TableMapperpublicabstractclassTableMapper<KEYOUT,VALUEOUT>extends

Mapper<ImmutableBytesWritable,Result,KEYOUT,VALUEOUT>{}輸入鍵類型，行鍵的字節(jié)數(shù)組輸入值類型，行鍵對應(yīng)的列數(shù)據(jù)思考提示輸出鍵/值類型TableMapper3-112/41自定義TableMapperstaticclassMyMapperextendsTableMapper<Text,Text>{

protectedvoidmap(ImmutableBytesWritablekey,Resultvalue, Contextcontext)throwsIOException,InterruptedException{ List<Cell>cells=value.listCells();

for(Cellcell:cells){

StringoutValue=String.format("RowKey:%sFamily:%sQualifier:%scellValue:%s", Bytes.toString(key.get()),//行鍵

Bytes.toString(CellUtil.cloneFamily(cell)),//列族

Bytes.toString(CellUtil.cloneQualifier(cell)),//列修飾符

Bytes.toString(CellUtil.cloneValue(cell))//單元格值

);

context.write(newText(CellUtil.getCellKeyAsString(cell)),

newText(outValue));

}

}}示例TableMapper3-213/41運行TableMapperConfigurationconf=HBaseConfiguration.create();conf.set(TableInputFormat.INPUT_TABLE,"music");conf.set(TableInputFormat.SCAN_COLUMNS,"info:nameinfo:gender");Jobjob=Job.getInstance(conf,"hbase-mapreduce-api");//MapReduce程序作業(yè)基本配置job.setJarByClass(TableMapperDemo.class);//通過驅(qū)動類的jar文件查找job.setInputFormatClass(TableInputFormat.class);//輸入格式j(luò)ob.setMapperClass(MyMapper.class);job.setMapOutputKeyClass(Text.class);//Mapper輸出鍵類型job.setMapOutputValueClass(Text.class);//Mapper輸出值類型……job.getConfiguration().set("tmpjars",StringUtils.join(libjars,','));……示例演示示例1：使用TableMapper處理數(shù)據(jù)指定被讀取的表指定被掃描的列解決任務(wù)對jar文件的依賴TableMapper3-314/41TableInputFormat常用常量設(shè)置TableInputFormat與HBaseScan對象常量值在任務(wù)運行時初始化為Scan的相關(guān)參數(shù)常量說明INPUT_TABLE指定表名SCAN_COLUMN_FAMILY指定列族SCAN_COLUMNS指定列，如：”info:nameinfo:gender”SCAN_ROW_START指定掃描行鍵范圍的起始行SCAN_ROW_STOP指定掃描行鍵范圍的結(jié)束行SCAN_TIMERANGE_START指定時間范圍的起始時間SCAN_TIMERANGE_END指定時間范圍的結(jié)束時間15/41第三方j(luò)ar文件依賴3-1客戶端執(zhí)行階段類路徑作業(yè)jar文件作業(yè)jar文件lib和classes$HADOOP_CLASSPATH客戶端執(zhí)行階段16/41第三方j(luò)ar文件依賴3-2任務(wù)執(zhí)行階段類路徑Hadoop執(zhí)行基本MapReduce程序的基礎(chǔ)類庫路徑作業(yè)jar文件作業(yè)jar文件中的lib和classes由-libjars選項指定的jar文件分布式緩存（DistributedCache）中的所有文件任務(wù)執(zhí)行階段注意不是$HADOOP_CLASSPATH17/41解決方案將依賴庫追加至客戶端$HADOOP_CLASSPATH中通過-libjars選項將依賴庫添加到任務(wù)的類路徑第三方j(luò)ar文件依賴3-3提示可使用GenericOptionsParser將“-libjars”選項值賦給Configuration的“tmpjars”屬性!18/41TableMapReduceUtil簡化Job配置的輔助工具類簡化第三方依賴處理addDependencyJars(Configuration,Class<?>...)addDependencyJars(Job)簡化Job初始化initTableMapperJob()initTableReducerJob()演示：簡化Job配置Scanscan=newScan();scan.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"));scan.addColumn(Bytes.toBytes("info"),Bytes.toBytes("gender"));TableMapReduceUtil.initTableMapperJob("music",scan,MyMapper.class, Text.class,Text.class,job);示例集成了addDependencyJars(job)19/41HBase輸出格式TableMapper是否支持輸出到HBase呢？TableOutputFormat繼承OutputFormat<KEY,Mutation>輸出鍵可為任意類型，而輸出值只能是Mutation使用要點指定TableMapper的KEYOUT、VALUEOUT指定Job的輸出格式為TableOutputFormat設(shè)置輸出表：TableOutputFormat.OUTPUT_TABLE問題提示必須是Mutation類型演示示例2：輸出歌曲清單Put、Get、Delete操作的父類20/41學員操作——使用TableMapper需求說明掃描music表的列info:name和info:terminal，并分別輸出到HDFS文件和HBase中，HDFS文件：

HBase表名：namelist，列族：details實現(xiàn)思路創(chuàng)建MapReduce工程添加HBase依賴庫編寫客戶端程序編寫map任務(wù)處理程序配置作業(yè)運行作業(yè)指導提示完成時間：25分鐘RowKey:[行鍵]

Family:[列族] Qualifier:[列修飾符]Value:[單元值]21/41常見問題及解決辦法代碼規(guī)范問題調(diào)試技巧共性問題集中講解共性問題集中講解22/41TableReducer3-1類定義TableReducer的輸出值類型VALUEOUT是什么？TableOutputFormat的輸出值類型是MutationTableReducer以TableOutputFormat作為輸出格式TableReducer的處理結(jié)果能輸出到HDFS嗎？publicabstractclassTableReducer<KEYIN,VALUEIN,KEYOUT>extendsReducer<KEYIN,VALUEIN,KEYOUT,Mutation>｛｝問題提示思考23/41自定義TableReducerTableReducer3-2staticclassMyReducerextendsTableReducer<Text,IntWritable,Text>{

@Override protectedvoidreduce(Textkey,Iterable<IntWritable>values, Contextcontext)throwsIOException,InterruptedException{ intplayCount=0; for(IntWritablenum:values){ playCount+=num.get(); } //為Put操作指定行鍵

Putput=newPut(Bytes.toBytes(key.toString())); //為Put操作指定列和值

put.addColumn(Bytes.toBytes("details"),Bytes.toBytes("rank"), Bytes.toBytes(playCount)); context.write(key,put); }}示例接收來自Mapper的輸出鍵/值類型輸出鍵類型輸出值類型24/41運行TableReducerTableReducer3-3Configurationconf=HBaseConfiguration.create();GenericOptionsParsergop=newGenericOptionsParser(conf,args);Jobjob=Job.getInstance(conf,"top-music");job.setJarByClass(TableReducerDemo.class); job.setNumReduceTasks(2);Scanscan=newScan();scan.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"));TableMapReduceUtil.initTableMapperJob("music",scan,MyMapper.class, Text.class,IntWritable.class,job);TableMapReduceUtil.initTableReducerJob("namelist",MyReducer.class,job);示例演示示例3：使用TableReducer25/41學員操作——使用TableReducer需求說明使用TableMapper掃描“music”表對每一次播放記錄中的歌曲名稱進行計數(shù)將計數(shù)結(jié)果輸出到TableReducer進行匯總將匯總結(jié)果輸出到表“namelist”實現(xiàn)思路編寫map任務(wù)程序編寫reduce任務(wù)程序配置作業(yè)運行客戶端指導提示完成時間：25分鐘26/41常見問題及解決辦法代碼規(guī)范問題調(diào)試技巧共性問題集中講解共性問題集中講解27/41小結(jié)比較之前的MapReduce實現(xiàn)步驟map處理使用TableMapperreduce處理使用TableReducerJob配置使用TableMapReduceUtil工具類初始化作業(yè)輸入/輸出格式輸入格式：TableInputFormat輸出格式：TableOutputFormat28/41實現(xiàn)音樂排行榜3-1運行流程29/41實現(xiàn)音樂排行榜3-2HBase表結(jié)構(gòu)行鍵時間戳列族(info)列健值1_song1_2016-1-1112:03:01t5info:namesong1t4info:singersinger1t3info:gendermant2info:rythmeslowt1info:terminalpc……行鍵時間戳列族(details)列健值song1t2details:rank3song2t1details:rank5……music表namelist表30/41實現(xiàn)音樂排行榜3-3程序?qū)崿F(xiàn)統(tǒng)計作業(yè)ScanMusicMapper

：掃描music表每一行中的列info:nameIntNumReducer

：匯總單曲播放總次數(shù)并輸出到namelist排序作業(yè)ScanMusicNameMapper

：掃描全部歌曲名稱，獲得單曲播放次數(shù)，并輸出HDFS文件IntWritableDecreaseingComparator

：實現(xiàn)降序顯示結(jié)果showResult()：讀取HDFS文件map處理時將鍵和值互換，并使用默認Reducer輸出演示示例4：音樂排行榜31/41HBase表設(shè)計優(yōu)化列族數(shù)量越少越好避免使用時序或單調(diào)遞增（遞減）的行鍵盡量最小化行鍵和列族的大小版本的數(shù)量不要設(shè)置過大32/41全局共享數(shù)據(jù)的方法讀寫HDFS文件占用較多系統(tǒng)資源配置Job屬性適合少量的共享數(shù)據(jù)使用分布式緩存適合大量的共享數(shù)據(jù)，但是只讀33/41使用分布式緩存復制文件至HDFS配置作業(yè)使用緩存文件Configurationconf=newConfiguration();Jobjob=Job.getInstance(conf,

“job1”);job.addCacheFile(newURI(“/input2/file/file1.txt#myfile”));job.addFileToClassPath(newPath(“/input2/file/a.jar”));示例publicstaticclassMyMapperextendsMapper<Object,Text,Text,IntWritable>{ publicvoidmap