基于Hadoop生态系统的大数据解决方案TP05

第五章Hadoop综合应用预习检查请说出TableMapper与Mapper的区别请说出TableReducer与Reducer的区别如何解决客户端和任务的第三方依赖库问题？在Job运行期间，共享全局变量有哪些方法？集中测试2/41回顾与作业点评简述HMaster、HRegionServer、HRegion间的关系描述HBase概念视图的结构HBaseShell常用操作有哪些？HBase数据存储管理API有哪些？点评作业的提交情况和共性问题回顾作业点评3/41本章任务使用HDFS+MapReduce+HBase实现音乐排行榜4/41本章目标理解HBase和MapReduce的集成使用掌握快速导入数据到HBase掌握MapReduce接收HBase数据输入掌握MapReduce输出到HBase5/41MapReduce与HBase集成策略使用MapReduce计算使用HBase存储、查询数据优势对HBase中的数据进行非实时性的统计分析对HBase中的数据进行分布式计算在多个MapReduce间使用HBase作为存储媒介方法使用HBaseMapReduceAPI6/41HBaseMapReduceAPI简介位于HBase提供的“hbase-server-1.0.2.jar”继承自HadoopMapReduceAPI需在Hadoop集群中配置HBase的环境复制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中的环境配置7/41HBaseMapReduce应用示例“hbase-server-1.0.2.jar”提供的应用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结合使用实现批量数据导入HDFS目录，表示数据来源演示行数统计和批量导入8/41学员操作——HBase批量导入需求说明获取素材music1.txt、music2.txt、music3.txt使用批量导入工具导入音乐播放记录，其中表名：music行键格式：ID_NAME_DATE列族：info列键：info:name，info:singer，info:gender，info:ryghme，info:terminal实现思路将hbase-site.xml复制到Hadoop配置目录将HBase类库中的jar文件添加到$HADOOP_CLASSPATH环境变量中运行hbase-server-1.0.2.jar中的importtsv和completebulkload作业导入数据完成时间：20分钟练习9/41常见问题及解决办法代码规范问题调试技巧共性问题集中讲解共性问题集中讲解10/41编写HBaseMapReduce依赖的jar文件HBase

jar文件MapReducejar文件MapReduce编程模型MapperReducerJob涉及到的类TableMapperTableInputFormatTableOutputFormatTableReducerTableMapReduceUtil11/41类定义为什么TableMapper不提供输入键/值类型？TableInputFormat继承InputFormat<ImmutableBytesWritable,Result>读取HBase行数据并封装为<ImmutableBytesWritable,Result>并传递给TableMapperpublicabstractclassTableMapper<KEYOUT,VALUEOUT>extends

Mapper<ImmutableBytesWritable,Result,KEYOUT,VALUEOUT>{}输入键类型，行键的字节数组输入值类型，行键对应的列数据思考提示输出键/值类型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程序作业基本配置job.setJarByClass(TableMapperDemo.class);//通过驱动类的jar文件查找job.setInputFormatClass(TableInputFormat.class);//输入格式job.setMapperClass(MyMapper.class);job.setMapOutputKeyClass(Text.class);//Mapper输出键类型job.setMapOutputValueClass(Text.class);//Mapper输出值类型……job.getConfiguration().set("tmpjars",StringUtils.join(libjars,','));……示例演示示例1：使用TableMapper处理数据指定被读取的表指定被扫描的列解决任务对jar文件的依赖TableMapper3-314/41TableInputFormat常用常量设置TableInputFormat与HBaseScan对象常量值在任务运行时初始化为Scan的相关参数常量说明INPUT_TABLE指定表名SCAN_COLUMN_FAMILY指定列族SCAN_COLUMNS指定列，如：”info:nameinfo:gender”SCAN_ROW_START指定扫描行键范围的起始行SCAN_ROW_STOP指定扫描行键范围的结束行SCAN_TIMERANGE_START指定时间范围的起始时间SCAN_TIMERANGE_END指定时间范围的结束时间15/41第三方jar文件依赖3-1客户端执行阶段类路径作业jar文件作业jar文件lib和classes$HADOOP_CLASSPATH客户端执行阶段16/41第三方jar文件依赖3-2任务执行阶段类路径Hadoop执行基本MapReduce程序的基础类库路径作业jar文件作业jar文件中的lib和classes由-libjars选项指定的jar文件分布式缓存（DistributedCache）中的所有文件任务执行阶段注意不是$HADOOP_CLASSPATH17/41解决方案将依赖库追加至客户端$HADOOP_CLASSPATH中通过-libjars选项将依赖库添加到任务的类路径第三方jar文件依赖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设置输出表：TableOutputFormat.OUTPUT_TABLE问题提示必须是Mutation类型演示示例2：输出歌曲清单Put、Get、Delete操作的父类20/41学员操作——使用TableMapper需求说明扫描music表的列info:name和info:terminal，并分别输出到HDFS文件和HBase中，HDFS文件：

HBase表名：namelist，列族：details实现思路创建MapReduce工程添加HBase依赖库编写客户端程序编写map任务处理程序配置作业运行作业指导提示完成时间：25分钟RowKey:[行键]

Family:[列族] Qualifier:[列修饰符]Value:[单元值]21/41常见问题及解决办法代码规范问题调试技巧共性问题集中讲解共性问题集中讲解22/41TableReducer3-1类定义TableReducer的输出值类型VALUEOUT是什么？TableOutputFormat的输出值类型是MutationTableReducer以TableOutputFormat作为输出格式TableReducer的处理结果能输出到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”表对每一次播放记录中的歌曲名称进行计数将计数结果输出到TableReducer进行汇总将汇总结果输出到表“namelist”实现思路编写map任务程序编写reduce任务程序配置作业运行客户端指导提示完成时间：25分钟26/41常见问题及解决办法代码规范问题调试技巧共性问题集中讲解共性问题集中讲解27/41小结比较之前的MapReduce实现步骤map处理使用TableMapperreduce处理使用TableReducerJob配置使用TableMapReduceUtil工具类初始化作业输入/输出格式输入格式：TableInputFormat输出格式：TableOutputFormat28/41实现音乐排行榜3-1运行流程29/41实现音乐排行榜3-2HBase表结构行键时间戳列族(info)列健值1_song1_2016-1-1112:03:01t5info:namesong1t4info:singersinger1t3info:gendermant2info:rythmeslowt1info:terminalpc……行键时间戳列族(details)列健值song1t2details:rank3song2t1details:rank5……music表namelist表30/41实现音乐排行榜3-3程序实现统计作业ScanMusicMapper

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

：汇总单曲播放总次数并输出到namelist排序作业ScanMusicNameMapper

：扫描全部歌曲名称，获得单曲播放次数，并输出HDFS文件IntWritableDecreaseingComparator

：实现降序显示结果showResult()：读取HDFS文件map处理时将键和值互换，并使用默认Reducer输出演示示例4：音乐排行榜31/41HBase表设计优化列族数量越少越好避免使用时序或单调递增（递减）的行键尽量最小化行键和列族的大小版本的数量不要设置过大32/41全局共享数据的方法读写HDFS文件占用较多系统资源配置Job属性适合少量的共享数据使用分布式缓存适合大量的共享数据，但是只读33/41使用分布式缓存复制文件至HDFS配置作业使用缓存文件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