MapReduce將文本數(shù)據(jù)導(dǎo)入到HBase中

1. 整體描述：將本地文件的數(shù)據(jù)整理之后導(dǎo)入到hbase中
   

2. 在HBase中創(chuàng)建表

   
   

3. 數(shù)據(jù)格式

   
   

4. MapReduce程序
   

   
   

   map程序

   
   

   package com.hadoop.mapreduce.test.map;
   
   import java.io.IOException;
   
   import org.apache.hadoop.io.Text;
   import org.apache.hadoop.mapreduce.Mapper;
   
   public class WordCountHBaseMapper extends Mapper<Object, Text, Text, Text>{
       
       public Text keyValue = new Text();
       public Text valueValue = new Text();
       //數(shù)據(jù)類(lèi)型為：key@addressValue#ageValue#sexValue
       @Override
       protected void map(Object key, Text value, Context context)
               throws IOException, InterruptedException {
           String lineValue = value.toString();
           
           if(lineValue != null){
               String[] valuesArray = lineValue.split("@");
               context.write(new Text(valuesArray[0]), new Text(valuesArray[1]));
           }
       }
   }

   Reduce程序

   
   

   package com.hadoop.mapreduce.test.reduce;
   
   import java.io.IOException;
   import java.util.Iterator;
   
   import org.apache.hadoop.hbase.client.Put;
   import org.apache.hadoop.hbase.mapreduce.TableReducer;
   import org.apache.hadoop.io.NullWritable;
   import org.apache.hadoop.io.Text;
   
   public class WordCountHBaseReduce extends TableReducer<Text, Text, NullWritable>{
   
       @Override
       protected void reduce(Text key, Iterable<Text> value, Context out)
               throws IOException, InterruptedException {
           String keyValue = key.toString();
           Iterator<Text> valueIterator = value.iterator();
           while(valueIterator.hasNext()){
               Text valueV = valueIterator.next();
               String[] valueArray = valueV.toString().split("#");
               
               Put putRow = new Put(keyValue.getBytes());
               putRow.add("address".getBytes(), "baseAddress".getBytes(), 
                           valueArray[0].getBytes());
               putRow.add("sex".getBytes(), "baseSex".getBytes(), 
                           valueArray[1].getBytes());
               putRow.add("age".getBytes(), "baseAge".getBytes(), 
                           valueArray[2].getBytes());
               
               out.write(NullWritable.get(), putRow);
           }
       }
   }

   主程序

   
   

   
   

   package com.hadoop.mapreduce.test;
   
   import java.io.IOException;
   
   import org.apache.hadoop.conf.Configuration;
   import org.apache.hadoop.fs.Path;
   import org.apache.hadoop.hbase.HBaseConfiguration;
   import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
   import org.apache.hadoop.io.Text;
   import org.apache.hadoop.mapreduce.Job;
   import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
   
   import com.hadoop.mapreduce.test.map.WordCountHBaseMapper;
   import com.hadoop.mapreduce.test.reduce.WordCountHBaseReduce;
   
   /**
    * 將hdfs上的內(nèi)容讀取到，并插入到hbase的表中，然后讀取hbase表中的內(nèi)容，將統(tǒng)計(jì)結(jié)果插入到hbase中 
    */
   public class WordCountHBase {
       public static void main(String args[]) throws IOException, 
           InterruptedException, ClassNotFoundException{
           
           Configuration conf = HBaseConfiguration.create();
           conf.set("hbase.zookeeper.quorum", "192.168.192.137"); 
           Job job = Job.getInstance(conf, "MapReduceHbaseJob");
           //各種class
           job.setJarByClass(WordCountHBase.class);
           job.setMapperClass(WordCountHBaseMapper.class);
           TableMapReduceUtil.initTableReducerJob("userInfo3", 
                  WordCountHBaseReduce.class, job);
           
           FileInputFormat.addInputPath(job, new Path(args[0]));
           job.setMapOutputKeyClass(Text.class);
           job.setMapOutputValueClass(Text.class);
           
           System.exit(job.waitForCompletion(true) ? 0 : 1);
       }
   }

   結(jié)果：

   
   

5. 注：如果運(yùn)行的client沒(méi)有hbase，需要在hadoop里面的lib中加入hbase的lib