自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡

MapReduce操作HBase錯誤一例

作者：Liqizhou
數(shù)據(jù)庫 大數(shù)據(jù)
操作HBase時經(jīng)常會遇到MasterNotRunningException的錯誤，那我們該如何處理這樣的錯誤呢？

運行HBase時常會遇到個錯誤，我就有這樣的經(jīng)歷。

ERROR: org.apache.hadoop.hbase.MasterNotRunningException: Retried 7 times

檢查日志：org.apache.hadoop.ipc.RPC$VersionMismatch: Protocol org.apache.hadoop.hdfs.protocol.ClientProtocol version mismatch. (client = 42, server = 41)

如果是這個錯誤，說明RPC協(xié)議不一致所造成的，解決方法：將hbase/lib目錄下的hadoop-core的jar文件刪除，將hadoop目錄下的hadoop-0.20.2-core.jar拷貝到hbase/lib下面，然后重新啟動hbase即可。第二種錯誤是：沒有啟動hadoop，先啟用hadoop，再啟用hbase。

在Eclipse開發(fā)中，需要加入hadoop所有的jar包以及HBase二個jar包（hbase，zooKooper）。

HBase基礎(chǔ)可見帖子：http://www.cnblogs.com/liqizhou/archive/2012/05/14/2499112.html

建表，通過HBaseAdmin類中的create靜態(tài)方法來創(chuàng)建表。

HTable類是操作表，例如，靜態(tài)方法put可以插入數(shù)據(jù)，該類初始化時可以傳遞一個行鍵，靜態(tài)方法getScanner()可以獲得某一列上的所有數(shù)據(jù)，返回Result類,Result類中有個靜態(tài)方法getFamilyMap（）可以獲得以列名為key，值為value，這剛好與hadoop中map結(jié)果是一樣的。

  1. package test;  
  2. import java.io.IOException;  
  3. import java.util.Map;  
  4. import org.apache.hadoop.conf.Configuration;  
  5. import org.apache.hadoop.hbase.HBaseConfiguration;  
  6. import org.apache.hadoop.hbase.HColumnDescriptor;  
  7. import org.apache.hadoop.hbase.HTableDescriptor;  
  8. import org.apache.hadoop.hbase.client.HBaseAdmin;  
  9. import org.apache.hadoop.hbase.client.HTable;  
  10. import org.apache.hadoop.hbase.client.Put;  
  11. import org.apache.hadoop.hbase.client.Result;  
  12.  
  13. public class Htable {  
  14.  
  15.     /**  
  16.      * @param args  
  17.      */ 
  18.     public static void main(String[] args) throws IOException {  
  19.         // TODO Auto-generated method stub  
  20.         Configuration hbaseConf = HBaseConfiguration.create();  
  21.         HBaseAdmin admin = new HBaseAdmin(hbaseConf);  
  22.         HTableDescriptor htableDescriptor = new HTableDescriptor("table" 
  23.                 .getBytes());  //set the name of table  
  24.         htableDescriptor.addFamily(new HColumnDescriptor("fam1")); //set the name of column clusters  
  25.         admin.createTable(htableDescriptor); //create a table   
  26.         HTable table = new HTable(hbaseConf, "table"); //get instance of table.  
  27.         for (int i = 0; i < 3; i++) {   //for is number of rows  
  28.             Put putRow = new Put(("row" + i).getBytes()); //the ith row  
  29.             putRow.add("fam1".getBytes(), "col1".getBytes(), "vaule1" 
  30.                     .getBytes());  //set the name of column and value.  
  31.             putRow.add("fam1".getBytes(), "col2".getBytes(), "vaule2" 
  32.                     .getBytes());  
  33.             putRow.add("fam1".getBytes(), "col3".getBytes(), "vaule3" 
  34.                     .getBytes());  
  35.             table.put(putRow);  
  36.         }  
  37.         for(Result result: table.getScanner("fam1".getBytes())){//get data of column clusters   
  38.             for(Map.Entry<byte[], byte[]> entry : result.getFamilyMap("fam1".getBytes()).entrySet()){//get collection of result  
  39.                 String column = new String(entry.getKey());  
  40.                 String value = new String(entry.getValue());  
  41.                 System.out.println(column+","+value);  
  42.             }  
  43.         }  
  44.         admin.disableTable("table".getBytes()); //disable the table  
  45.         admin.deleteTable("table".getBytes());  //drop the tbale  
  46.     }  
  47. } 

以上代碼不難看懂。

下面介紹一下，用mapreduce怎樣操作HBase，主要對HBase中的數(shù)據(jù)進行讀取。

現(xiàn)在有一些大的文件，需要存入HBase中，其思想是先把文件傳到HDFS上，利用map階段讀取<key,value>對，可在reduce把這些鍵值對上傳到HBase中。

  1. package test;  
  2. import java.io.IOException;  
  3. import java.util.Map;  
  4. import org.apache.hadoop.conf.Configuration;  
  5. import org.apache.hadoop.hbase.HBaseConfiguration;  
  6. import org.apache.hadoop.hbase.HColumnDescriptor;  
  7. import org.apache.hadoop.hbase.HTableDescriptor;  
  8. import org.apache.hadoop.hbase.client.HBaseAdmin;  
  9. import org.apache.hadoop.hbase.client.HTable;  
  10. import org.apache.hadoop.hbase.client.Put;  
  11. import org.apache.hadoop.hbase.client.Result;  
  12.  
  13. public class Htable {  
  14.  
  15.     /**  
  16.      * @param args  
  17.      */ 
  18.     public static void main(String[] args) throws IOException {  
  19.         // TODO Auto-generated method stub  
  20.         Configuration hbaseConf = HBaseConfiguration.create();  
  21.         HBaseAdmin admin = new HBaseAdmin(hbaseConf);  
  22.         HTableDescriptor htableDescriptor = new HTableDescriptor("table" 
  23.                 .getBytes());  //set the name of table  
  24.         htableDescriptor.addFamily(new HColumnDescriptor("fam1")); //set the name of column clusters  
  25.         admin.createTable(htableDescriptor); //create a table   
  26.         HTable table = new HTable(hbaseConf, "table"); //get instance of table.  
  27.         for (int i = 0; i < 3; i++) {   //for is number of rows  
  28.             Put putRow = new Put(("row" + i).getBytes()); //the ith row  
  29.             putRow.add("fam1".getBytes(), "col1".getBytes(), "vaule1" 
  30.                     .getBytes());  //set the name of column and value.  
  31.             putRow.add("fam1".getBytes(), "col2".getBytes(), "vaule2" 
  32.                     .getBytes());  
  33.             putRow.add("fam1".getBytes(), "col3".getBytes(), "vaule3" 
  34.                     .getBytes());  
  35.             table.put(putRow);  
  36.         }  
  37.         for(Result result: table.getScanner("fam1".getBytes())){//get data of column clusters   
  38.             for(Map.Entry<byte[], byte[]> entry : result.getFamilyMap("fam1".getBytes()).entrySet()){//get collection of result  
  39.                 String column = new String(entry.getKey());  
  40.                 String value = new String(entry.getValue());  
  41.                 System.out.println(column+","+value);  
  42.             }  
  43.         }  
  44.         admin.disableTable("table".getBytes()); //disable the table  
  45.         admin.deleteTable("table".getBytes());  //drop the tbale  
  46.     }  
  47. } 

Reduce類,主要是將鍵值傳到HBase表中

  1. package test;  
  2.  
  3. import java.io.IOException;  
  4. import org.apache.hadoop.hbase.client.Put;  
  5. import org.apache.hadoop.hbase.io.ImmutableBytesWritable;  
  6. import org.apache.hadoop.hbase.mapreduce.TableReducer;  
  7. import org.apache.hadoop.io.Text;  
  8.  
  9. public class ReducerClass extends TableReducer<Text,Text,ImmutableBytesWritable>{  
  10.     public void reduce(Text key,Iterable<Text> values,Context context){  
  11.         String k = key.toString();  
  12.         StringBuffer str=null;  
  13.         for(Text value: values){  
  14.             str.append(value.toString());  
  15.         }  
  16.         String v = new String(str);   
  17.         Put putrow = new Put(k.getBytes());  
  18.         putrow.add("fam1".getBytes(), "name".getBytes(), v.getBytes());       
  19.     }  
  20. } 

由上面可知ReducerClass繼承TableReduce，在hadoop里面ReducerClass繼承Reducer類。它的原型為：TableReducer<KeyIn,Values,KeyOut>可以看出，HBase里面是讀出的Key類型是ImmutableBytesWritable。

Map，Reduce，以及Job的配置分離，比較清晰,mahout也是采用這種構(gòu)架。

  1. package test;  
  2.  
  3. import org.apache.hadoop.conf.Configuration;  
  4. import org.apache.hadoop.conf.Configured;  
  5. import org.apache.hadoop.fs.Path;  
  6. import org.apache.hadoop.hbase.HBaseConfiguration;  
  7. import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;  
  8. import org.apache.hadoop.io.Text;  
  9. import org.apache.hadoop.mapreduce.Job;  
  10. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
  11. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;  
  12. import org.apache.hadoop.util.Tool;  
  13.  
  14. public class Driver extends Configured implements Tool{  
  15.  
  16.     @Override 
  17.     public static void run(String[] arg0) throws Exception {  
  18.         // TODO Auto-generated method stub  
  19.         Configuration conf = HBaseConfiguration.create();  
  20.         conf.set("hbase.zookeeper.quorum.", "localhost");    
  21.           
  22.         Job job = new Job(conf,"Hbase");  
  23.         job.setJarByClass(TxtHbase.class);  
  24.           
  25.         Path in = new Path(arg0[0]);  
  26.           
  27.         job.setInputFormatClass(TextInputFormat.class);  
  28.         FileInputFormat.addInputPath(job, in);  
  29.           
  30.         job.setMapperClass(MapperClass.class);  
  31.         job.setMapOutputKeyClass(Text.class);  
  32.         job.setMapOutputValueClass(Text.class);  
  33.           
  34.         TableMapReduceUtil.initTableReducerJob("table", ReducerClass.class, job);  
  35.           
  36.        job.waitForCompletion(true);  
  37.     }  
  38.       
  39. } 

Driver中job配置的時候沒有設(shè)置 job.setReduceClass(); 而是用 TableMapReduceUtil.initTableReducerJob("tab1", THReducer.class, job); 來執(zhí)行reduce類。

主函數(shù)

  1. package test;  
  2.  
  3. import org.apache.hadoop.conf.Configuration;  
  4. import org.apache.hadoop.util.ToolRunner;  
  5.  
  6. public class TxtHbase {  
  7.     public static void main(String [] args) throws Exception{        Driver.run(new Configuration(),new THDriver(),args);     } } 

讀取數(shù)據(jù)時比較簡單，編寫Mapper函數(shù)，讀取<key,value>值就行了。

  1. package test;  
  2.  
  3. import java.io.IOException;  
  4. import org.apache.hadoop.conf.Configuration;  
  5. import org.apache.hadoop.hbase.client.Result;  
  6. import org.apache.hadoop.hbase.io.ImmutableBytesWritable;  
  7. import org.apache.hadoop.hbase.mapred.TableMap;  
  8. import org.apache.hadoop.io.Text;  
  9. import org.apache.hadoop.mapred.MapReduceBase;  
  10. import org.apache.hadoop.mapred.OutputCollector;  
  11. import org.apache.hadoop.mapred.Reporter;  
  12.  
  13. public class MapperClass extends MapReduceBase implements 
  14.         TableMap<Text, Text> {  
  15.     static final String NAME = "GetDataFromHbaseTest";  
  16.     private Configuration conf;  
  17.  
  18.     public void map(ImmutableBytesWritable row, Result values,  
  19.             OutputCollector<Text, Text> output, Reporter reporter)  
  20.             throws IOException {  
  21.         StringBuilder sb = new StringBuilder();  
  22.         for (Entry<byte[], byte[]> value : values.getFamilyMap(  
  23.                 "fam1".getBytes()).entrySet()) {  
  24.             String cell = value.getValue().toString();  
  25.             if (cell != null) {  
  26.                 sb.append(new String(value.getKey())).append(new String(cell));  
  27.             }  
  28.         }  
  29.         output.collect(new Text(row.get()), new Text(sb.toString()));  
  30.     } 

要實現(xiàn)這個方法 initTableMapJob(String table, String columns, Class<? extends TableMap> mapper, Class<? extends org.apache.hadoop.io.WritableComparable> outputKeyClass, Class<? extends org.apache.hadoop.io.Writable> outputValueClass, org.apache.hadoop.mapred.JobConf job, boolean addDependencyJars)。

  1. package test;  
  2.  
  3. import org.apache.hadoop.conf.Configuration;  
  4. import org.apache.hadoop.conf.Configured;  
  5. import org.apache.hadoop.fs.Path;  
  6. import org.apache.hadoop.hbase.HBaseConfiguration;  
  7. import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;  
  8. import org.apache.hadoop.io.Text;  
  9. import org.apache.hadoop.mapreduce.Job;  
  10. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
  11. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;  
  12. import org.apache.hadoop.util.Tool;  
  13.  
  14. public class Driver extends Configured implements Tool{  
  15.  
  16.     @Override 
  17.     public static void run(String[] arg0) throws Exception {  
  18.         // TODO Auto-generated method stub  
  19.         Configuration conf = HBaseConfiguration.create();  
  20.         conf.set("hbase.zookeeper.quorum.", "localhost");    
  21.         Job job = new Job(conf,"Hbase");  
  22.         job.setJarByClass(TxtHbase.class);  
  23.         job.setInputFormatClass(TextInputFormat.class);  
  24.         job.setMapOutputKeyClass(Text.class);  
  25.         job.setMapOutputValueClass(Text.class);  
  26.         TableMapReduceUtilinitTableMapperJob("table", args0[0],MapperClass.class, job);         job.waitForCompletion(true); } } 

主函數(shù)

  1. package test;  
  2.  
  3. import org.apache.hadoop.conf.Configuration;  
  4. import org.apache.hadoop.util.ToolRunner;  
  5.  
  6. public class TxtHbase {  
  7.     public static void main(String [] args) throws Exception{  
  8.  
  9.         Driver.run(new Configuration(),new THDriver(),args);  
  10.  
  11.     }   
  12. }  
  13. -------------------------------------------------------------------------------- 

原文鏈接：http://www.cnblogs.com/liqizhou/archive/2012/05/17/2504279.html

【編輯推薦】

  1. 數(shù)據(jù)庫遷移之何去何從
  2. SQL Server數(shù)據(jù)庫遷移偏方
  3. SQL Server數(shù)據(jù)庫恢復(fù)案例分享
  4. SQL Server數(shù)據(jù)庫最小宕機遷移方案
  5. 給你大型數(shù)據(jù)庫遷移的五大建議  
     
責任編輯：彭凡 來源： 博客園
MapReduce
相關(guān)推薦

2014-11-19 09:22:48

云計算Dockerpython API

2012-09-24 01:01:49

NginxNginx性能Web服務(wù)器

2009-07-16 13:03:05

ibatis resu

2013-11-12 14:43:43

MySQL數(shù)據(jù)庫

2009-12-01 09:15:30

Windows 7系統(tǒng)激活

2009-08-07 09:57:38

2009-09-17 16:38:02

WSUS服務(wù)器

2017-01-16 15:43:54

存儲虛擬化控制器

2013-10-15 09:48:03

C++Lambda函數(shù)式編程

2021-03-08 06:29:53

微信僵尸粉移動應(yīng)用

2021-07-06 12:07:27

Go 服務(wù)性能

2020-11-10 13:42:07

Go編譯器修復(fù)

2009-08-06 11:37:24

虛擬機NAT連接物理網(wǎng)絡(luò)

2022-12-26 09:05:35

2017-12-08 11:28:48

Hbase木桶效應(yīng)操作

2019-09-09 08:30:57

MYSQL代碼數(shù)據(jù)庫

2010-06-03 13:55:38

Hbase和Hadoo

2011-03-01 09:43:13

MapReduce架構(gòu)

2019-04-18 13:00:13

2009-12-28 09:58:52

ADO操作
點贊
收藏

51CTO技術(shù)棧公眾號