hbase-.-程式設計-大数据文档资料

王耀聰陳威宇●HBase程式編譯方法●HBase程式設計◆常用的HBaseAPI說明◆實做I/O操作◆搭配MapReduce運算●其他用法補充●其他專案2程式編譯方法Method1-使用JavaJDK1.6Method2-使用Eclipse套件1.Java之編譯與執行hadoop_home/lib/資料夾內2.編譯◆javacΔ-classpathΔhadoop-*-core.jar:hbase-*.jarΔ-dΔMyJavaΔMyCode.java◆jarΔ-cvfΔMyJar.jarΔ-CΔMyJavaΔ.4.執行◆bin/hadoopΔjarΔMyJar.jarΔMyCodeΔ{Input/ΔOutput/}‧./MyJava‧./MyJava=編譯後程式碼目錄‧Myjar.jar=封裝後的編譯檔‧./input;./ouput不一定為hdf的s輸42.Eclipse之編譯與執行5HBase程式設計常用的HBaseAPI說明實做I/O操作搭配MapReduce運算HBase程式設計常用的HBaseAP明●●●●Table,FamilyColumn,QualifierRow,TimeStamp,Cell,LockHBase常用函式●●●●●●●HBaseAdminHBaseConfigurationDatabaseHTableHTableDescriptor}-TableFamilyColumnQualifier9●AddsHBase◆=newHBaseConfiguration()◆=newHBaseConfiguration(Configurationc)●繼承自回傳值●HBas的e管理介面◆=newHBaseAdmin(HBaseConfigurationconf)●Ex:HBaseAdminHBaseAdminadmin=newHBaseAdmadmin.disableTable(“tablename”);回傳值(StringtableName,HColumnDesc(HBaseConfigurationconf)HTableDescriptor[](byte[]tableName,HTableDboolean●HTableDescriptorcontainsthenameofanHTable,anditscolumnfamilies.◆=newHTableDescriptor()◆=newHTableDescriptor(Stringname)●Constant-values◆org.apache.hadoop.hbase.HTableDescriptor.TABLE_DESCRIPTOR_VERSION●Ex:HTableDescriptorhtd=newHTableDesHTableDescriptorhtd=newHTableDeshtd.addFamily(newHColumnDescrip回傳值byte[]byte[]●AnHColumnDescriptorcontainsinformationaboutacolumnfamily◆=newHColumnDescriptor(Stringfamilyname)●Constant-values◆org.apache.hadoop.hbase.HTableDescriptor.TABLE_DESCRIPTOR_VERSION●Ex:HTableDescriptorhtd=newHTableDescHTableDescriptorhtd=newHTableDescHColumnDescriptorcol=newHColumnDescri回傳值byte[]byte[]●UsedtocommunicatewithasingleHBasetable.◆=newHTable(HBaseConfigurationconf,StringtableName)●Ex:HTabletable=newHTable(conf,Bytes.toByHTabletable=newHTable(conf,Bytes.toByResultScannerscanner=tab回傳值(byte[]row,byte[]family,byte[]qualibooleanbyte[][]byte[]put●UsedtoperformPutoperationsforasinglerow.◆=newPut(byte[]row)◆=newPut(byte[]row,RowLockrowLock)●Ex:HTabletable=newHTable(conf,Bytes.toByHTabletable=newHTable(conf,Bytes.toByp.add(family,qualibyte[]boolean●UsedtoperformGetoperationsonasinglerow.◆=newGet(byte[]row)◆=newGet(byte[]row,RowLockrowLock)●Ex:HTableHTabletable=newHTable(conf,Bytes●AlloperationsareidenticaltoGet◆Ratherthanspecifyingasinglerow,anoptionalstartRowandstopRow●Ifrowsarenotspecified,theScannerwilliterateoverallrows.◆=newScan()◆=newScan(byte[]startRow,byte[]stopRow)◆=newScan(byte[]startRow,Filterfilter)●SinglerowresultofaGetorScanquery.◆=newResult()●Ex:HTableHTabletable=newHTable(conf,BytesbooleanNavigableMapbyte[]byte[]●Interfaceforclient-sidescanning.GotoHTabletoobtaininstances.◆HTable.getScanner(Bytes.toBytes(family));●Ex:ResultScannerscanner=table.getScannResultScannerscanner=table.getScann}voidcloseResultnext●org.apache.hadoop.hbase.KeyValue●getRow(),getFamily(),getQualifier(),getTimestamp(),andgetValue().●TheKeyValueblobformatinsidethebytearrayis:<keylength><valuelengt◆Key的格式:◆<row-family-type>◆Rowlength最大值為Short.MAX_SIZE,◆columnfamilylength最大值為Byte.MAX_SIZE,◆columnqualifier+keylength必須小於Integer.MAX_SIZE.HBase程式設計實做I/O操作範例一：新增Table範例一：新增TablepublicstaticvoidcreateHBaseTable(Stringtablenam{HTableDescriptorhtd=newHTableDescriptor(tabHColumnDescriptorcol=newHColumnDescriptor("cHBaseConfigurationconfig=newHBaseConfiguHBaseAdminadmin=newHBaseAdminif(admin.tableExist{}}範例二：Pu料進Column範例二：Pu料進ColumnHBaseConfigurationconfig=newHBaseConHTabletable=newHTable(confbyte[]brow=Bytes.toBytbyte[]bfamily=Bytes.tbyte[]bcolumn=Bytes.toBytebyte[]bvalue=Bytes.toBytes(p.add(bfamily,bcolumn,}範例三：GetColumnValue範例三：GetColumnValuestaticStringgetColumn(Stringtablename,Stringrow,Stringfamily,Stringcolumn){HBaseConfigurationconf=newHBaseConfiguration();Stringret="";HTabletable;try{table=newHTable(conf,Bytes.toBytes(tablename));Getg=newGet(Bytes.toBytes(row));ResultrowResult=table.get(g);ret=Bytes.toString(rowResult.getValue(Bytes.toBytes(family+“:”+column)));table.close();}catch(IOExceptionee.printStackTrace();}returnret;}範例四：ScanallColumn範例四：ScanallColumnstaticvoidScanColumn(Stringtablename,Stringfamily,Stringcolumn){HBaseConfigurationconf=newHBaseConfiguration();HTabletable;try{table=newHTable(conf,Bytes.toBytes(tablename));ResultScannerscanner=table.getScanner(Bytes.toBytes(family));for(ResultrowResult:scanner){byte[]by=rowResult.getValue(Bytes.toBytes(family),Bytes.toBytes(column));Stringstr=Bytes.toString(by);System.out.println("row"+i+"is\""+str+"\"");i++;}}catch(IOExceptionee.printStackTrace();}範例五：刪除資料表範例五：刪除資料表staticvoiddrop(Stringtablename){HBaseConfigurationconf=newHBaseConfiguration();try{HBaseAdminadmin=newHBaseAdmin(conf);if(admin.tableExists(tablename)){admin.disableTable(tablename);admin.deleteTable(tablename);System.out.println("Dropedthetable["+tablename+"]");System.out.println("Table["+tablename+"]wasnotfound!");}}catch(IOExceptionee.printStackTrace();}HBase程式設計範例六：WordCountHBase●程式說明範例六：WordCountHBasepublicclassWordCountHBase{publicstaticclassMapextenMapper<LongWritable,Text,Te{privateIntWritablei=newpublicvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,Interrupt{{context.write(newTe}}publicstaticclassReduceextTableReducer<Text,IntWrNullWritable>{publicvoidreduce(Textkey,Iterable<IntWritable>values,context)throwsIOE{{}Put(Bytes.toBytes(key.toStrput.add(Bytes.toBytes("content"),Bytes.toBytes("count"),Bytes.toBytes(String.valueOf(context.write(NullWritable.ge}範例六：WordCountHBasepublicstaticvoidcreateHBaseTable(Stringtablename)throwsIOExcepti{HTableDescriptorhtd=HTableDescriptor(tablenaHColumnDescriptorcol=newHColumnDescriptor("content:"htd.addFamily(col);HBaseConfigurationconfig=newHBaseAdminadmin=nif(admin.tableExists(tabl{admin.disableTable(tableadmin.deleteTable(tabl}System.out.println("create}}publicstaticvoidmain(Stringargs[])throwsException{Configurationconf=newConfiguration();conf.set(TableOutputFormat.OUTPUT_TABLtablename);createHBaseTable(tablename);Jobjob=newJob(conf,"WordCounttablewjob.setJarByClass(WordCountHBase.class);job.setNumReduceTasks(3);job.setMapperClass(Map.class);job.setReducerClass(Reduce.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);job.setInputFormatClass(TextInputFormat.class);job.setOutputFormatClass(TableOutputFormat.class);FileInputFormat.addInputPath(job,newPath(input));System.exit(job.waitForCompletion(true)?0:1);}範例六範例七：LoadHBaseMapper說明：此程式碼將HBas的e資料取出來，再將結果塞回hdfs上運算方法：將此程式運作在hadoop0.20平台上，用(參考2的)方法加入hbase參數後，將此程式碼打包成XX.jar執行：hadoopjarXX.jarLoadHBaseMapper<hdfs_output>---------------------------結果：$hadoopfs-cat<hdfs_output>/part-r-00000---------------------------GunLongEsingSunDonStarBucks---------------------------1.請注意hbase上必須要有table,並且已經有資料2.運算完後，程式將執行結果放在你指定hdf的s<hdfs_output>內請注意沒有<hdfs_output>資料夾範例七：LoadHBaseMapperpublicclassLoadHBaseMapper{publicstaticclassHtMapextendsTableMapper<Text,Text>{publicvoidmap(ImmutableBytesWritablekey,Resultvalue,Contextcontext)throwsIOException,InterruptedException{Stringres=Bytes.toString(value.getValue(Bytes.toBytes("Detail"),Bytes.toBytes("Name")));context.write(newText(key.toString()),newText(res));publicstaticclassHtReduceextendsReducer<Text,Text,Text,Text>{publicvoidreduce(Textkey,Iterable<Text>values,Contextcontext)throwsIOException,InterruptedException{Stringstr=newString("");Textfinal_key=newText(key);Textfinal_value=newText();for(Texttmp:values){str+=tmp.toString();}final_value.set(str);context.write(final_key,final_value);範例七：LoadHBaseMapperpublicstaticvoidmain(Stringargs[])throwsException{Stringinput=args[0];Stringtablename="tsmc";Configurationconf=newConfiguration();Jobjob=newJob(conf,tablename+"hbasedatatohdfs");job.setJarByClass(LoadHBaseMapper.class);TableMapReduceUtil.initTableMapperJob(tablename,myScan,HtMap.class,Text.class,Text.class,job);job.setMapperClass(HtMap.class);job.setReducerClass(HtReduce.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(Text.class);job.setInputFormatClass(TableInputFormat.clasjob.setOutputFormatClass(TextOutputFormat.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(Text.class);FileOutputFormat.setOutputPath(job,newPath(input));System.exit(job.waitForCompletion(true)?0:範例七Thrift●IndexedTable=SecondaryIndex=TransactionalHBase●內容與原本table相似的另一張table，但key不同，利於排列內容2254813banana1253banana481.環境設定增加兩項內容<name>hbase.regionserver.class<<value>org.apache.hadoop.hbase.ipc.IndexedRegionInterface</va<name>hbase.regionserver.impl<org.apache.hadoop.hbase.regionserver.tableindexed.IndexedRegio