Hadoop生态圈-HBase的HFile创建方式

　　　　　　　　　　　　　　　　　　　　　　Hadoop生态圈-HBase的HFile创建方式

　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　作者：尹正杰

版权声明：原创作品，谢绝转载！否则将追究法律责任。

　　废话不多说，直接上代码，想说的话都在代码的注释里面。

一.环境准备

list

create 'yinzhengjie:WordCount3','f1','f2'

list

desc 'yinzhengjie:WordCount3'

scan 'yinzhengjie:WordCount3'

二.编写HFile创建方式的代码

1>.编写Map端代码

 /*

 @author :yinzhengjie

 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E7%94%9F%E6%80%81%E5%9C%88/

 EMAIL:[email protected]

 */

 package cn.org.yinzhengjie.hbase.hfile;

 import org.apache.hadoop.io.IntWritable;

 import org.apache.hadoop.io.LongWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Mapper;

 import java.io.IOException;

 public class HFileOutputMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

     @Override

     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

         //得到一行数据

         String line = value.toString();

         String[] arr = line.split(" ");

         //

         for (String word : arr){

             context.write(new Text(word),new IntWritable(1));

         }

     }

 }

2>.编写Reducer端代码

 /*

 @author :yinzhengjie

 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E7%94%9F%E6%80%81%E5%9C%88/

 EMAIL:[email protected]

 */

 package cn.org.yinzhengjie.hbase.hfile;

 import org.apache.hadoop.hbase.Cell;

 import org.apache.hadoop.hbase.CellUtil;

 import org.apache.hadoop.hbase.KeyValue;

 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

 import org.apache.hadoop.hbase.util.Bytes;

 import org.apache.hadoop.io.IntWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Reducer;

 import java.io.IOException;

 public class HFileOutputReducer extends Reducer<Text,IntWritable,ImmutableBytesWritable,Cell> {

     @Override

     protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

         int sum = 0;

         for (IntWritable value : values) {

             sum += value.get();

         }

         if(key.toString().length() > 0){

             ImmutableBytesWritable outKey = new ImmutableBytesWritable(Bytes.toBytes(key.toString()));

             //创建cell

             Cell cell = CellUtil.createCell(Bytes.toBytes(key.toString()),

                     Bytes.toBytes("f1"), Bytes.toBytes("count"),System.currentTimeMillis(),

                     KeyValue.Type.Minimum,Bytes.toBytes(sum+""),null);

             context.write(outKey,cell);

         }

     }

 }

3>.编写主程序代码

 /*

 @author :yinzhengjie

 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E7%94%9F%E6%80%81%E5%9C%88/

 EMAIL:[email protected]

 */

 package cn.org.yinzhengjie.hbase.hfile;

 import org.apache.hadoop.conf.Configuration;

 import org.apache.hadoop.fs.Path;

 import org.apache.hadoop.hbase.Cell;

 import org.apache.hadoop.hbase.HBaseConfiguration;

 import org.apache.hadoop.hbase.HTableDescriptor;

 import org.apache.hadoop.hbase.TableName;

 import org.apache.hadoop.hbase.client.Connection;

 import org.apache.hadoop.hbase.client.ConnectionFactory;

 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

 import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;

 import org.apache.hadoop.io.IntWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Job;

 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

 public class App {

     public static void main(String[] args) throws Exception {

         System.setProperty("HADOOP_USER_NAME", "yinzhengjie");

         Configuration conf = HBaseConfiguration.create();

         conf.set("fs.defaultFS","file:///");

         Connection conn = ConnectionFactory.createConnection(conf);

         Job job = Job.getInstance(conf);

         job.setJobName("HFile WordCount");

         job.setJarByClass(App.class);

         job.setMapperClass(HFileOutputMapper.class);

         job.setReducerClass(HFileOutputReducer.class);

         //设置输出格式

         job.setOutputFormatClass(HFileOutputFormat2.class);

         //设置路径

         FileInputFormat.addInputPath(job,new Path("file:///D:\\BigData\\yinzhengjieData\\word.txt"));

         FileOutputFormat.setOutputPath(job,new Path("file:///D:\\BigData\\yinzhengjieData\\hfile"));

         //设置输出k-v

         job.setOutputKeyClass(ImmutableBytesWritable.class);

         job.setOutputValueClass(Cell.class);

         //设置map端输出k-v

         job.setMapOutputKeyClass(Text.class);

         job.setMapOutputValueClass(IntWritable.class);

         /**

          *      配置和"yinzhengjie:WordCount3"进行关联，也就是说"yinzhengjie:WordCount3"这个表必须在HBase数据库中存在，

          * 实际操作是以"yinzhengjie:WordCount3"为模板，便于生成HFile文件！

          */

         HFileOutputFormat2.configureIncrementalLoad(job, new HTableDescriptor(TableName.valueOf("yinzhengjie:WordCount3")),

                 conn.getRegionLocator(TableName.valueOf("yinzhengjie:WordCount3")) );

         job.waitForCompletion(true);

     }

 }

4>.查看测试结果