Chapter05基于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输出到HBase,5/41,MapReduce与HBase集成,策略使用MapReduce计算使用HBase存储、查询数据优势对HBase中的数据进行非实时性的统计分析对HBase中的数据进行分布式计算在多个MapReduce间使用HBase作为存储媒介方法使用HBaseMapReduceAPI,6/41,HBaseMapReduceAPI简介,位于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/*,7/41,HBaseMapReduce应用示例,“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.jarimporttsvDimporttsv.bulk.output=tmpDimporttsv.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.jarcompletebulkloadtmpmusic,HBase表名,指定生成的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作业导入数据,练习,9/41,常见问题及解决办法代码规范问题调试技巧,共性问题集中讲解,10/41,编写HBaseMapReduce,依赖的jar文件HBasejar文件MapReducejar文件MapReduce编程模型MapperReducerJob涉及到的类TableMapperTableInputFormatTableOutputFormatTableReducerTableMapReduceUtil,11/41,类定义为什么TableMapper不提供输入键/值类型？TableInputFormat继承InputFormat读取HBase行数据并封装为并传递给TableMapper,publicabstractclassTableMapperextendsMapper,输入键类型，行键的字节数组,输入值类型，行键对应的列数据,思考,提示,输出键/值类型,TableMapper3-1,12/41,自定义TableMapper,staticclassMyMapperextendsTableMapperprotectedvoidmap(ImmutableBytesWritablekey,Resultvalue,Contextcontext)throwsIOException,InterruptedExceptionListcells=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-2,13/41,运行TableMapper,Configurationconf=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,);,示例,指定被读取的表,指定被扫描的列,解决任务对jar文件的依赖,TableMapper3-3,14/41,TableInputFormat,常用常量设置TableInputFormat与HBaseScan对象常量值在任务运行时初始化为Scan的相关参数,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_CLASSPATH,17/41,解决方案将依赖库追加至客户端$HADOOP_CLASSPATH中通过-libjars选项将依赖库添加到任务的类路径,第三方jar文件依赖3-3,提示,18/41,TableMapReduceUtil,简化Job配置的辅助工具类简化第三方依赖处理addDependencyJars(Configuration,Class.)addDependencyJars(Job)简化Job初始化initTableMapperJob()initTableReducerJob(),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/41,HBase输出格式,TableMapper是否支持输出到HBase呢？TableOutputFormat继承OutputFormat输出键可为任意类型，而输出值只能是Mutation使用要点指定TableMapper的KEYOUT、VALUEOUT指定Job的输出格式为TableOutputFormat设置输出表：TableOutputFormat.OUTPUT_TABLE,问题,提示,必须是Mutation类型,Put、Get、Delete操作的父类,20/41,学员操作使用TableMapper,需求说明扫描music表的列info:name和info:terminal，并分别输出到HDFS文件和HBase中，HDFS文件：HBase表名：namelist，列族：details实现思路创建MapReduce工程添加HBase依赖库编写客户端程序编写map任务处理程序配置作业运行作业,指导,提示,RowKey:行键Family:列族Qualifier:列修饰符Value:单元值,21/41,常见问题及解决办法代码规范问题调试技巧,共性问题集中讲解,22/41,TableReducer3-1,类定义TableReducer的输出值类型VALUEOUT是什么？TableOutputFormat的输出值类型是MutationTableReducer以TableOutputFormat作为输出格式TableReducer的处理结果能输出到HDFS吗？,publicabstractclassTableReducerextendsReducer,问题,提示,思考,23/41,自定义TableReducer,TableReducer3-2,staticclassMyReducerextendsTableReducerOverrideprotectedvoidreduce(Textkey,Iterablevalues,Contextcontext)throwsIOException,InterruptedExceptionintplayCount=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,运行TableReducer,TableReducer3-3,Configurationconf=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);,示例,25/41,学员操作使用TableReducer,需求说明使用TableMapper扫描“music”表对每一次播放记录中的歌曲名称进行计数将计数结果输出到TableReducer进行汇总将汇总结果输出到表“namelist”实现思路编写map任务程序编写reduce任务程序配置作业运行客户端,指导,提示,26/41,常见问题及解决办法代码规范问题调试技巧,共性问题集中讲解,27/41,小结,比较之前的MapReduce实现步骤map处理使用TableMapperreduce处理使用TableReducerJob配置使用TableMapReduceUtil工具类初始化作业输入/输出格式输入格式：TableInputFormat输出格式：TableOutputFormat,28/41,实现音乐排行榜3-1,运行流程,29/41,实现音乐排行榜3-2,HBase表结构,music表,namelist表,30/41,实现音乐排行榜3-3,程序实现统计作业ScanMusicMapper：扫描music表每一行中的列info:nameIntNumReducer：汇总单曲播放总次数并输出到namelist排序作业ScanMusicNameMapper：扫描全部歌曲名称，获得单曲播放次数，并输出HDFS文件IntWritableDecreaseingComparator：实现降序显示结果showResult()：读取HDFS文件,map处理时将键和值互换，并使用默认Reducer输出,31/41,HBase表设计优化,列族数量越少越好避免使用时序或单调递增（递减）的行键尽量最小化行键和列族的大小版本的数量不要设置过大,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”);,示例,publicstaticclassMyMapperextendsMapperpublicvoidmap(Objectkey,Textvalue,Contextcontext)FileReaderreader=newFileReader(“myf