2023學(xué)年完整公開課版Hbase批量處理

Hbase批量處理課程內(nèi)容Hbase基礎(chǔ)HbaseShellHbaseJava編程Hbase過(guò)濾器Hbase批量處理HBase與MapReduce在偽分布式模式和完全分布式模式下的HBase是架構(gòu)在HDFS之上的，因此完全可以將MapReduce編程框架和HBase結(jié)合起來(lái)使用。將HBase作為底層存儲(chǔ)結(jié)構(gòu)，MapReduce調(diào)用HBase進(jìn)行特殊的處理，這樣能夠充分結(jié)合HBase分布式大型數(shù)據(jù)庫(kù)和MapReduce并行計(jì)算的優(yōu)點(diǎn)。HBase實(shí)現(xiàn)了TableInputFormatBase類，該類提供了對(duì)表數(shù)據(jù)的大部分操作，其子類TableInputFormat則提供了完整的實(shí)現(xiàn)，用于處理表數(shù)據(jù)并生成鍵值對(duì)。TableInputFormat類將數(shù)據(jù)表按照Region分割成split，即有多少個(gè)Regions就有多個(gè)splits，然后將Region按行鍵分成<key,value>對(duì)，key值對(duì)應(yīng)行鍵，value值為該行所包含的數(shù)據(jù)。HBase實(shí)現(xiàn)了MapReduce計(jì)算框架對(duì)應(yīng)的TableMapper類和TableReducer類。HBase實(shí)現(xiàn)的TableOutputFormat將輸出的<key,value>對(duì)寫到指定的HBase表中。HBase與MapReduce將MapReduce和HBase結(jié)合起來(lái)的WordCount程序，從指定文件中搜集數(shù)據(jù)，進(jìn)行統(tǒng)計(jì)計(jì)算，最后將結(jié)果存儲(chǔ)到HBase中。publicstaticclasshBaseReducerextendsTableReducer<Text,IntWritable,NullWritable>{ protectedvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{intsum=0; for(IntWritablevalue:values){ sum+=value.get(); } Putput=newPut(key.toString().getBytes());

put.addColumn("content".getBytes(),"count".getBytes(), String.valueOf(sum).getBytes());

context.write(NullWritable.get(),put); }}HBase的Bulkload如果HDFS中有海量數(shù)據(jù)要導(dǎo)入HBase，通常MapReduce在寫Hbase時(shí)使用的是TableOutputFormat

方式，在reduce中直接生成put對(duì)象寫入hbase，該方式在大數(shù)據(jù)量寫入時(shí)效率低下，并對(duì)HBase節(jié)點(diǎn)的穩(wěn)定性造成一定的影響。HBase支持bulkload的入庫(kù)方式，它是利用hbase的數(shù)據(jù)信息按照特定格式存儲(chǔ)在hdfs內(nèi)這一原理，直接在HDFS中生成持久化的HFile數(shù)據(jù)格式文件，然后上傳至合適位置，即完成巨量數(shù)據(jù)快速入庫(kù)的辦法。配合mapreduce完成，高效便捷，而且不占用region資源，增添負(fù)載，在大數(shù)據(jù)量寫入時(shí)能極大的提高寫入效率，并降低對(duì)HBase節(jié)點(diǎn)的寫入壓力。

HBase的BulkloadMap中主要代碼：publicstaticclassbulkMapperextendsMapper<LongWritable,Text,ImmutableBytesWritable,Put>{ protectedvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{ ... ImmutableBytesWritablerowKeyWritable=new ImmutableBytesWritable(rowKey); Putput=newPut(rowKey); ... context.write(rowKeyWritable,put); }}HBase的BulkloadMain中主要代碼：…job.setMapperClass(bulkMapper.class); job.setReducerClass(PutSortReducer.class); job.setMapOutputKeyClass(ImmutableBytesWritable.class);job.setMapOutputValueClass(Put.class);job.setOutputFormatClass(HFileOutputFormat2.class);...HBase的BulkloadConnectionconn=ConnectionFactory.createConnection(conf);TableName

tn=TableName.valueOf("test");HTabletable=(HTable)conn.getTable(tn);

RegionLocator

regionLocator=conn.getRegionLocator(tn);HFileOutputFormat2.configureIncrementalLoad(job,table,regionLocator);job.waitForCompletion(true);LoadIncrementalHFilesloader=new