使用MapReduce处理Hbase数据

ゝ一纸荒年。 2022-09-18 13:45 228阅读 0赞

[使用MapReduce处理Hbase数据][MapReduce_Hbase]

今天终于把MR处理Hbase的数据的程序搞定了，自己走了好多的弯路，程序写完之后，在本机的伪分布式的hadoop上跑是没问题的，可是把程序上传的集群上就出错了，最后发现是zookeeper没配对，在编译的时候没有把conf添加的CLASSPATH，这才导致出错的。

下面是MR测试的程序：

![复制代码][copycode.gif]

1 import java.io.IOException;
     2 
     3 import org.apache.hadoop.conf.Configuration;
     4 import org.apache.hadoop.hbase.HBaseConfiguration;
     5 import org.apache.hadoop.hbase.HColumnDescriptor;
     6 import org.apache.hadoop.hbase.HTableDescriptor;
     7 import org.apache.hadoop.hbase.client.HBaseAdmin;
     8 import org.apache.hadoop.hbase.client.Put;
     9 import org.apache.hadoop.hbase.client.Result;
    10 import org.apache.hadoop.hbase.client.Scan;
    11 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
    12 import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
    13 import org.apache.hadoop.hbase.mapreduce.TableMapper;
    14 import org.apache.hadoop.hbase.mapreduce.TableReducer;
    15 import org.apache.hadoop.hbase.util.Bytes;
    16 import org.apache.hadoop.io.IntWritable;
    17 import org.apache.hadoop.io.Text;
    18 import org.apache.hadoop.mapreduce.Job;
    19 
    20 public class Test {
    21     private static final String sourceTable = "sourceTable";
    22     private static final String targetTable = "targetTable";
    23     static Configuration config = HBaseConfiguration.create();
    24     
    25     public static void createTable(String tablename, String[] cfs) throws IOException {
    26         HBaseAdmin admin = new HBaseAdmin(config);
    27         if (admin.tableExists(tablename)) {
    28             System.out.println("table already exists");
    29         }
    30         else {
    31             HTableDescriptor tableDesc = new HTableDescriptor(tablename);
    32             for (int i = 0; i < cfs.length; i++) {
    33                 tableDesc.addFamily(new HColumnDescriptor(cfs[i]));
    34             }
    35             admin.createTable(tableDesc);
    36             System.out.println("create table successly");
    37         }
    38     }
    39     /**
    40      * @param args
    41      * @throws IOException
    42      * @throws ClassNotFoundException 
    43      * @throws InterruptedException 
    44      */
    45     public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
    46         // TODO Auto-generated method stub
    47         String[] cfs = {"a"};
    48         createTable(targetTable, cfs);
    49         Job job = new Job(config, "test");
    50         job.setJarByClass(Test.class);
    51         Scan scan = new Scan();
    52         scan.setCaching(1024);
    53         scan.setCacheBlocks(false);
    54         TableMapReduceUtil.initTableMapperJob(
    55                 sourceTable,        
    56                 scan,               
    57                 Mapper1.class,    
    58                 Text.class,         
    59                 IntWritable.class,  
    60                 job);    
    61         TableMapReduceUtil.initTableReducerJob(
    62                 targetTable,        
    63                 Reducer1.class,    
    64                 job);
    65         boolean b = job.waitForCompletion(true);
    66         if(!b){
    67             throw new IOException("error");
    68         }
    69     }
    70 
    71     public static class Mapper1 extends
    72             TableMapper<Text, IntWritable> {
    73             private final IntWritable ONE = new IntWritable(1);
    74             private Text text = new Text();
    75             public void map(ImmutableBytesWritable row, Result value, Context context) throws IOException, InterruptedException{
    76                 String id = new String(value.getValue(Bytes.toBytes("cf"), Bytes.toBytes("a")));
    77                 text.set(id);
    78                     context.write(id, ONE);
    79             }
    80     }
    81     public static class Reducer1 extends TableReducer<Text, IntWritable, ImmutableBytesWritable>{
    82         public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException{
    83             int i = 0;
    84             for (IntWritable val : values){
    85                 i += val.get();
    86             }
    87             Put put = new Put(Bytes.toBytes(key.toString()));
    88             put.add(Bytes.toBytes("a"), Bytes.toBytes("c"), Bytes.toBytes(i));
    89             context.write(null, put);
    90         }
    91     }
    92 }

![复制代码][copycode.gif]

编写完成后需要打包，打包可以在本地打，也可以在服务器上的包，一定要设置CLASSPATH

export CLASSPATH = /data/hadoop/hadoop-1.0.4/hadoop-core-1.0.4.jar:/data/hadoop/hbase-0.94.2/hbase-0.94.2.jar:/data/hadoop/hbase-0.94.2/conf/

在终端运行这个命令或者直接将此命令下载家目录下的.bashrc中也可以，

然后创建  test\_classes文件夹，

运行命令：

javac -d test\_classes/ Test.java

运行完成后会在test\_classes文件夹下生成3个.class文件

然后运行

jar -cvf test.jar -C test\_classes .

即可生成test.jar 文件

最后运行:

bin/hadoop jar test.jar Test

运行MR程序即可

[MapReduce_Hbase]: http://www.cnblogs.com/hitandrew/archive/2013/01/10/2855631.html
[copycode.gif]: /images/20220828/ecc5bee0dee949d2b01cbffb900413fb.png