图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Hbase › 其它› HBase 5种写入数据方式
分享

HBase 5种写入数据方式

howtodown 2014-4-20 14:01:40 发表于 其它 [显示全部楼层] 只看大图 回帖奖励 阅读模式 关闭右栏 8 35936
About云VIP帮助找工作和提升
本帖最后由 howtodown 于 2014-4-20 14:04 编辑
问题导读：
1.如何直接使用HTable进行导入？
2.如何从HDFS文件导入HBase，继承自Mapper？
3.如何读取HBase表写入HBase表中字段？
4.如何让MR和HTable结合？






Version ：hadoop1.2.1; hbaes0.94.16;

HBase写入数据方式（参考：《HBase The Definitive Guide》），可以简单分为下面几种：

1. 直接使用HTable进行导入，代码如下：

  1. package hbase.curd;

  2. 

  3. import java.io.IOException;

  4. import java.util.ArrayList;

  5. import java.util.List;

  6. import java.util.Random;

  7. 

  8. import org.apache.hadoop.hbase.client.HTable;

  9. import org.apache.hadoop.hbase.client.Put;

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

  11. 

  12. public class PutExample {

  13. 

  14.         /**

  15.          * @param args

  16.          * @throws IOException 

  17.          */

  18.         private HTable  table = HTableUtil.getHTable("testtable");

  19.         public static void main(String[] args) throws IOException {

  20.                 // TODO Auto-generated method stub

  21.                 PutExample pe = new PutExample();

  22.                 pe.putRows();

  23.                 

  24.         }

  25.         

  26.         public void putRows(){

  27.                 List<Put> puts = new ArrayList<Put>();

  28.                 for(int i=0;i<10;i++){

  29.                         Put put = new Put(Bytes.toBytes("row_"+i));

  30.                         Random random = new Random();

  31.                         

  32.                         if(random.nextBoolean()){

  33.                                 put.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"), Bytes.toBytes("colfam1_qual1_value_"+i));

  34.                         }

  35.                         if(random.nextBoolean()){

  36.                                 put.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual2"), Bytes.toBytes("colfam1_qual1_value_"+i));

  37.                         }

  38.                         if(random.nextBoolean()){

  39.                                 put.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual3"), Bytes.toBytes("colfam1_qual1_value_"+i));

  40.                         }

  41.                         if(random.nextBoolean()){

  42.                                 put.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual4"), Bytes.toBytes("colfam1_qual1_value_"+i));        

  43.                         }

  44.                         if(random.nextBoolean()){

  45.                                 put.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual5"), Bytes.toBytes("colfam1_qual1_value_"+i));

  46.                         }

  47.                         puts.add(put);

  48.                 }

  49.                 try{

  50.                         table.put(puts);

  51.                         table.close();

  52.                 }catch(Exception e){

  53.                         e.printStackTrace();

  54.                         return ;

  55.                 }

  56.                 System.out.println("done put rows");

  57.         }

  58. 

  59. }

复制代码

其中HTableUtil如下：
  1. package hbase.curd;

  2. 

  3. import java.io.IOException;

  4. 

  5. import org.apache.hadoop.conf.Configuration;

  6. import org.apache.hadoop.hbase.HBaseConfiguration;

  7. import org.apache.hadoop.hbase.client.HTable;

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

  9. 

  10. public class HTableUtil {

  11.         private static HTable table;

  12.         private static Configuration conf;

  13.         

  14.         static{

  15.                 conf =HBaseConfiguration.create();

  16.                 conf.set("mapred.job.tracker", "hbase:9001");

  17.                 conf.set("fs.default.name", "hbase:9000");

  18.                 conf.set("hbase.zookeeper.quorum", "hbase");

  19.                 try {

  20.                         table = new HTable(conf,"testtable");

  21.                 } catch (IOException e) {

  22.                         // TODO Auto-generated catch block

  23.                         e.printStackTrace();

  24.                 }

  25.         }

  26.         public static Configuration getConf(){

  27.                 return conf;

  28.         }

  29.         public static HTable getHTable(String tablename){

  30.                 if(table==null){

  31.                         try {

  32.                                 table= new HTable(conf,tablename);

  33.                         } catch (IOException e) {

  34.                                 // TODO Auto-generated catch block

  35.                                 e.printStackTrace();

  36.                         } 

  37.                 }

  38.                 return table;

  39.         }

  40.         

  41.         public static  byte[] gB(String name){

  42.                 return Bytes.toBytes(name);

  43.         }

  44. }

复制代码

这一种是没有使用MR的，下面介绍的几种方式都是使用MR的。

2.1 从HDFS文件导入HBase，继承自Mapper，代码如下：

  1. package hbase.mr;

  2. 

  3. import java.io.IOException;

  4. 

  5. import hbase.curd.HTableUtil;

  6. 

  7. import org.apache.commons.cli.CommandLine;

  8. import org.apache.commons.cli.CommandLineParser;

  9. import org.apache.commons.cli.HelpFormatter;

  10. import org.apache.commons.cli.Option;

  11. import org.apache.commons.cli.Options;

  12. import org.apache.commons.cli.PosixParser;

  13. import org.apache.commons.codec.digest.DigestUtils;

  14. import org.apache.hadoop.conf.Configuration;

  15. import org.apache.hadoop.fs.Path;

  16. import org.apache.hadoop.hbase.KeyValue;

  17. import org.apache.hadoop.hbase.client.Put;

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

  19. import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;

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

  21. import org.apache.hadoop.io.LongWritable;

  22. import org.apache.hadoop.io.Text;

  23. import org.apache.hadoop.io.Writable;

  24. import org.apache.hadoop.mapreduce.Job;

  25. import org.apache.hadoop.mapreduce.Mapper;

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

  27. import org.apache.hadoop.util.GenericOptionsParser;

  28. 

  29. public class ImportFromFile {

  30. 

  31.         /**

  32.          * 从文件导入到HBase

  33.          * @param args

  34.          */

  35.         public static final String NAME="ImportFromFile";

  36.         public enum Counters{LINES}

  37.         

  38.         static class ImportMapper extends Mapper<LongWritable,Text,

  39.                 ImmutableBytesWritable,Writable>{

  40.                 private byte[] family =null;

  41.                 private byte[] qualifier = null;

  42.                 @Override

  43.                 protected void setup(Context cxt){

  44.                         String column = cxt.getConfiguration().get("conf.column");

  45.                         byte[][] colkey = KeyValue.parseColumn(Bytes.toBytes(column));

  46.                         family = colkey[0];

  47.                         if(colkey.length>1){

  48.                                 qualifier = colkey[1];

  49.                         }

  50.                 }

  51.                 @Override

  52.                 public void map(LongWritable offset,Text line,Context cxt){

  53.                         try{

  54.                                 String lineString= line.toString();

  55.                                 byte[] rowkey= DigestUtils.md5(lineString);

  56.                                 Put put = new Put(rowkey);

  57.                                 put.add(family,qualifier,Bytes.toBytes(lineString));

  58.                                 cxt.write(new ImmutableBytesWritable(rowkey), put);

  59.                                 cxt.getCounter(Counters.LINES).increment(1);

  60.                         }catch(Exception e){

  61.                                 e.printStackTrace();

  62.                         }

  63.                 }

  64.         }

  65.         private static CommandLine parseArgs(String[] args){

  66.                 Options options = new Options();

  67.                 Option o = new Option("t" ,"table",true,"table to import into (must exist)");

  68.                 o.setArgName("table-name");

  69.                 o.setRequired(true);

  70.                 options.addOption(o);

  71.                 

  72.                 o= new Option("c","column",true,"column to store row data into");

  73.                 o.setArgName("family:qualifier");

  74.                 o.setRequired(true);

  75.                 options.addOption(o);

  76.                 

  77.                 o = new Option("i", "input", true,

  78.                 "the directory or file to read from");

  79.                 o.setArgName("path-in-HDFS");

  80.                 o.setRequired(true);

  81.                 options.addOption(o);

  82.                 options.addOption("d", "debug", false, "switch on DEBUG log level");

  83.                 CommandLineParser parser = new PosixParser();

  84.                 CommandLine cmd = null;

  85.                 try {

  86.                         cmd = parser.parse(options, args);

  87.                 } catch (Exception e) {

  88.                         System.err.println("ERROR: " + e.getMessage() + "\n");

  89.                         HelpFormatter formatter = new HelpFormatter();

  90.                         formatter.printHelp(NAME + " ", options, true);

  91.                         System.exit(-1);

  92.                 }

  93.                 return cmd;

  94.         }

  95.         public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

  96.                 

  97.                 Configuration conf = HTableUtil.getConf();

  98.                 String[] otherArgs = new GenericOptionsParser(conf, initialArg()).getRemainingArgs(); 

  99.                 CommandLine cmd = parseArgs(otherArgs);

  100.                 String table = cmd.getOptionValue("t");

  101.                 String input = cmd.getOptionValue("i");

  102.                 String column = cmd.getOptionValue("c");

  103.                 conf.set("conf.column", column);

  104.                 Job job = new Job(conf, "Import from file " + input + " into table " + table);

  105.                 job.setJarByClass(ImportFromFile.class);

  106.                 job.setMapperClass(ImportMapper.class);

  107.                 job.setOutputFormatClass(TableOutputFormat.class);

  108.                 job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, table);

  109.                 job.setOutputKeyClass(ImmutableBytesWritable.class);

  110.                 job.setOutputValueClass(Writable.class);

  111.                 job.setNumReduceTasks(0); 

  112.                 FileInputFormat.addInputPath(job, new Path(input));

  113.                 System.exit(job.waitForCompletion(true) ? 0 : 1);

  114.         }

  115.         

  116.         private static String[] initialArg(){

  117.                 String []args = new String[6];

  118.                 args[0]="-c";

  119.                 args[1]="fam:data";

  120.                 args[2]="-i";

  121.                 args[3]="/user/hadoop/input/picdata";

  122.                 args[4]="-t";

  123.                 args[5]="testtable";

  124.                 return args;

  125.         }

  126. }

复制代码

2.2 读取HBase表写入HBase表中字段，代码如下：
  1. package hbase.mr;

  2. 

  3. import hadoop.util.HadoopUtils;

  4. 

  5. import java.io.IOException;

  6. 

  7. import org.apache.hadoop.conf.Configuration;

  8. import org.apache.hadoop.hbase.KeyValue;

  9. import org.apache.hadoop.hbase.client.Put;

  10. import org.apache.hadoop.hbase.client.Result;

  11. import org.apache.hadoop.hbase.client.Scan;

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

  13. import org.apache.hadoop.hbase.mapreduce.IdentityTableReducer;

  14. import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;

  15. import org.apache.hadoop.hbase.mapreduce.TableMapper;

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

  17. import org.apache.hadoop.io.Writable;

  18. import org.apache.hadoop.mapreduce.Job;

  19. import org.slf4j.Logger;

  20. import org.slf4j.LoggerFactory;

  21. 

  22. public class ParseDriver {

  23. 

  24.         /**

  25.          * 把hbase表中数据拷贝到其他表（或本表）相同字段

  26.          * @param args

  27.          */

  28.         enum Counters{

  29.                 VALID, ROWS, COLS, ERROR

  30.         }

  31.         private static Logger log = LoggerFactory.getLogger(ParseDriver.class);

  32.         static class ParseMapper extends TableMapper<ImmutableBytesWritable,Writable>{

  33.                 private byte[] columnFamily =null ;

  34.                 private byte[] columnQualifier =null;

  35.                 @Override

  36.                 protected void setup(Context cxt){

  37.                         columnFamily = Bytes.toBytes(cxt.getConfiguration().get("conf.columnfamily"));

  38.                         columnQualifier = Bytes.toBytes(cxt.getConfiguration().get("conf.columnqualifier"));

  39.                 }

  40.                 @Override 

  41.                 public void map(ImmutableBytesWritable row,Result columns,Context cxt){

  42.                         cxt.getCounter(Counters.ROWS).increment(1);

  43.                         String value =null;

  44.                         try{

  45.                                 Put put = new Put(row.get());

  46.                                 for(KeyValue kv : columns.list()){

  47.                                         cxt.getCounter(Counters.COLS).increment(1);

  48.                                         value= Bytes.toStringBinary(kv.getValue());

  49.                                         if(equals(columnQualifier,kv.getQualifier())){  // 过滤column

  50.                                                 put.add(columnFamily,columnQualifier,kv.getValue());

  51.                                                 cxt.write(row, put);

  52.                                                 cxt.getCounter(Counters.VALID).increment(1);

  53.                                         }

  54.                                 }

  55.                         }catch(Exception e){

  56.                                 log.info("Error:"+e.getMessage()+",Row:"+Bytes.toStringBinary(row.get())+

  57.                                                 ",Value:"+value);

  58.                                 cxt.getCounter(Counters.ERROR).increment(1);

  59.                         }

  60.                 }

  61.                 private boolean equals(byte[] a,byte[] b){

  62.                         String aStr= Bytes.toString(a);

  63.                         String bStr= Bytes.toString(b);

  64.                         if(aStr.equals(bStr)){

  65.                                 return true;

  66.                         }

  67.                         return false;

  68.                 }

  69.         }

  70.         

  71.         public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

  72.                 byte[] columnFamily = Bytes.toBytes("fam");

  73.                 byte[] columnQualifier = Bytes.toBytes("data");

  74.                 Scan scan = new Scan ();

  75.                 scan.addColumn(columnFamily, columnQualifier);

  76.                 HadoopUtils.initialConf("hbase");

  77.                 Configuration conf = HadoopUtils.getConf();

  78.                 conf.set("conf.columnfamily", Bytes.toStringBinary(columnFamily));

  79.                 conf.set("conf.columnqualifier", Bytes.toStringBinary(columnQualifier));

  80.                 

  81.                 String input ="testtable" ;//

  82.                 String output="testtable1"; // 

  83.                                 

  84.                 Job job = new Job(conf,"Parse data in "+input+",write to"+output);

  85.                 job.setJarByClass(ParseDriver.class);

  86.                 TableMapReduceUtil.initTableMapperJob(input, scan, ParseMapper.class, 

  87.                                 ImmutableBytesWritable.class, Put.class,job);

  88.                 TableMapReduceUtil.initTableReducerJob(output, IdentityTableReducer.class, job);

  89.                 

  90.                 System.exit(job.waitForCompletion(true)?0:1);

  91.                 

  92.         }

  93. 

  94. }

复制代码
其中HadoopUtils代码如下：
  1. package hadoop.util;

  2. 

  3. import java.io.IOException;

  4. import java.net.URI;

  5. import java.util.ArrayList;

  6. import java.util.List;

  7. 

  8. import org.apache.hadoop.conf.Configuration;

  9. import org.apache.hadoop.fs.FSDataInputStream;

  10. import org.apache.hadoop.fs.FileSystem;

  11. import org.apache.hadoop.fs.Path;

  12. import org.apache.hadoop.io.Text;

  13. import org.apache.hadoop.util.LineReader;

  14. 

  15. public class HadoopUtils {

  16. 

  17.         private static Configuration conf;

  18.         public  static void initialConf(){

  19.                 conf = new Configuration();

  20.                 conf.set("mapred.job.tracker", "hbase:9001");

  21.                 conf.set("fs.default.name", "hbase:9000");

  22.                 conf.set("hbase.zookeeper.quorum", "hbase");

  23.         }

  24.         public  static void initialConf(String host){

  25.                 conf = new Configuration();

  26.                 conf.set("mapred.job.tracker", host+":9001");

  27.                 conf.set("fs.default.name", host+":9000");

  28.                 conf.set("hbase.zookeeper.quorum", host);

  29.         }

  30.         public static Configuration getConf(){

  31.                 if(conf==null){

  32.                         initialConf();

  33.                 }

  34.                 return conf;

  35.         }

  36.         

  37.         public static List<String> readFromHDFS(String fileName) throws IOException {

  38.                 Configuration conf = getConf();

  39.                 FileSystem fs = FileSystem.get(URI.create(fileName), conf);

  40.                 FSDataInputStream hdfsInStream = fs.open(new Path(fileName));

  41.                 // 按行读取（新版本的方法）

  42.                 LineReader inLine = new LineReader(hdfsInStream, conf);

  43.                 Text txtLine = new Text();

  44.                 

  45.                 int iResult = inLine.readLine(txtLine); //读取第一行

  46.                 List<String> list = new ArrayList<String>();

  47.                 while (iResult > 0 ) {

  48.                         list.add(txtLine.toString());

  49.                         iResult = inLine.readLine(txtLine);

  50.                 }

  51.                 

  52.                 hdfsInStream.close();

  53.                 fs.close();

  54.                 return list;

  55.         }

  56. }

复制代码

2.3 MR和HTable结合，代码如下：
  1. package hbase.mr;

  2. 

  3. import hadoop.util.HadoopUtils;

  4. import hbase.mr.AnalyzeDriver.Counters;

  5. 

  6. import java.io.IOException;

  7. import java.util.Date;

  8. 

  9. import org.apache.hadoop.conf.Configuration;

  10. import org.apache.hadoop.hbase.KeyValue;

  11. import org.apache.hadoop.hbase.client.HTable;

  12. import org.apache.hadoop.hbase.client.Put;

  13. import org.apache.hadoop.hbase.client.Result;

  14. import org.apache.hadoop.hbase.client.Scan;

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

  16. import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;

  17. import org.apache.hadoop.hbase.mapreduce.TableMapper;

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

  19. import org.apache.hadoop.io.Writable;

  20. import org.apache.hadoop.mapreduce.Job;

  21. import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;

  22. import org.slf4j.Logger;

  23. import org.slf4j.LoggerFactory;

  24. 

  25. public class ParseSinglePutDriver {

  26. 

  27.         /**

  28.          * 使用HTable进行写入

  29.          * 把infoTable 表中的 qualifier字段复制到qualifier1字段

  30.          * 单个Put

  31.          * @param args

  32.          */

  33.         private static Logger log = LoggerFactory.getLogger(ParseMapper.class);

  34.         static class ParseMapper extends TableMapper<ImmutableBytesWritable,Writable>{

  35.                 private HTable infoTable =null ;

  36.                 private byte[] columnFamily =null ;

  37.                 private byte[] columnQualifier =null;

  38.                 private byte[] columnQualifier1 =null;

  39.                 @Override 

  40.                 protected void setup(Context cxt){

  41.                         log.info("ParseSinglePutDriver setup,current time: "+new Date());

  42.                         try {

  43.                                 infoTable = new HTable(cxt.getConfiguration(),

  44.                                                 cxt.getConfiguration().get("conf.infotable"));

  45.                                 infoTable.setAutoFlush(false);

  46.                         } catch (IOException e) {

  47.                                 log.error("Initial infoTable error:\n"+e.getMessage());

  48.                         }

  49.                         columnFamily = Bytes.toBytes(cxt.getConfiguration().get("conf.columnfamily"));

  50.                         columnQualifier = Bytes.toBytes(cxt.getConfiguration().get("conf.columnqualifier"));

  51.                         columnQualifier1 = Bytes.toBytes(cxt.getConfiguration().get("conf.columnqualifier1"));

  52.                 }

  53.                 @Override 

  54.                 protected void cleanup(Context cxt){

  55.                         try {

  56.                                 infoTable.flushCommits();

  57.                                 log.info("ParseSinglePutDriver cleanup ,current time :"+new Date());

  58.                         } catch (IOException e) {

  59.                                 log.error("infoTable flush commits error:\n"+e.getMessage());

  60.                         }

  61.                 }

  62.                 @Override 

  63.                 public void map(ImmutableBytesWritable row,Result columns,Context cxt){

  64.                         cxt.getCounter(Counters.ROWS).increment(1);

  65.                         String value =null ;

  66.                         try{

  67.                                 Put put = new Put(row.get());

  68.                                 for(KeyValue kv : columns.list()){

  69.                                         cxt.getCounter(Counters.COLS).increment(1);

  70.                                         value= Bytes.toStringBinary(kv.getValue());

  71.                                         if(equals(columnQualifier,kv.getQualifier())){  // 过滤column

  72.                                                 put.add(columnFamily,columnQualifier1,kv.getValue());

  73.                                                 infoTable.put(put);

  74.                                         }

  75.                                 }

  76.                         }catch(Exception e){

  77.                                 log.info("Error:"+e.getMessage()+",Row:"+Bytes.toStringBinary(row.get())+

  78.                                                 ",Value:"+value);

  79.                                 cxt.getCounter(Counters.ERROR).increment(1);

  80.                         }

  81.                 }

  82.                 private boolean equals(byte[] a,byte[] b){

  83.                         String aStr= Bytes.toString(a);

  84.                         String bStr= Bytes.toString(b);

  85.                         if(aStr.equals(bStr)){

  86.                                 return true;

  87.                         }

  88.                         return false;

  89.                 }

  90.         }

  91.         public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

  92.                 String input ="testtable";

  93.                 byte[] columnFamily = Bytes.toBytes("fam");

  94.                 byte[] columnQualifier = Bytes.toBytes("data");

  95.                 byte[] columnQualifier1 = Bytes.toBytes("data1");

  96.                 Scan scan = new Scan ();

  97.                 scan.addColumn(columnFamily, columnQualifier);

  98.                 HadoopUtils.initialConf("hbase");

  99.                 Configuration conf = HadoopUtils.getConf();

  100.                 conf.set("conf.columnfamily", Bytes.toStringBinary(columnFamily));

  101.                 conf.set("conf.columnqualifier", Bytes.toStringBinary(columnQualifier));

  102.                 conf.set("conf.columnqualifier1", Bytes.toStringBinary(columnQualifier1));

  103.                 conf.set("conf.infotable", input);

  104.                 

  105.                 Job job = new Job(conf,"Parse data in "+input+",into tables");

  106.                 job.setJarByClass(ParseSinglePutDriver.class);

  107.                 TableMapReduceUtil.initTableMapperJob(input, scan, ParseMapper.class, 

  108.                                 ImmutableBytesWritable.class, Put.class,job);        

  109.                 job.setOutputFormatClass(NullOutputFormat.class);

  110.                 job.setNumReduceTasks(0);

  111.                 System.exit(job.waitForCompletion(true)?0:1);

  112.         }

  113. 

  114. }

复制代码

2.4 上面2.3中的HTable其实也是可以put一个List的，下面的方式就是put一个list的方式，这样效率会高。
  1. package hbase.mr;

  2. 

  3. import hadoop.util.HadoopUtils;

  4. import hbase.mr.AnalyzeDriver.Counters;

  5. 

  6. import java.io.IOException;

  7. import java.util.ArrayList;

  8. import java.util.Date;

  9. import java.util.List;

  10. 

  11. import org.apache.hadoop.conf.Configuration;

  12. import org.apache.hadoop.hbase.KeyValue;

  13. import org.apache.hadoop.hbase.client.HTable;

  14. import org.apache.hadoop.hbase.client.Put;

  15. import org.apache.hadoop.hbase.client.Result;

  16. import org.apache.hadoop.hbase.client.Scan;

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

  18. import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;

  19. import org.apache.hadoop.hbase.mapreduce.TableMapper;

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

  21. import org.apache.hadoop.io.Writable;

  22. import org.apache.hadoop.mapreduce.Job;

  23. import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;

  24. import org.slf4j.Logger;

  25. import org.slf4j.LoggerFactory;

  26. 

  27. public class ParseListPutDriver {

  28. 

  29.         /**

  30.          * 使用HTable进行写入

  31.          * List <Put> 进行测试,查看效率

  32.          * 把infoTable 表中的 qualifier字段复制到qualifier1字段

  33.          * @param args

  34.          */

  35.         private static Logger log = LoggerFactory.getLogger(ParseMapper.class);

  36.         static class ParseMapper extends TableMapper<ImmutableBytesWritable,Writable>{

  37.                 private HTable infoTable =null ;

  38.                 private byte[] columnFamily =null ;

  39.                 private byte[] columnQualifier =null;

  40.                 private byte[] columnQualifier1 =null;

  41.                 private List<Put> list = new ArrayList<Put>();

  42.                 @Override 

  43.                 protected void setup(Context cxt){

  44.                         log.info("ParseListPutDriver setup,current time: "+new Date());

  45.                         try {

  46.                                 infoTable = new HTable(cxt.getConfiguration(),

  47.                                                 cxt.getConfiguration().get("conf.infotable"));

  48.                                 infoTable.setAutoFlush(false);

  49.                         } catch (IOException e) {

  50.                                 log.error("Initial infoTable error:\n"+e.getMessage());

  51.                         }

  52.                         columnFamily = Bytes.toBytes(cxt.getConfiguration().get("conf.columnfamily"));

  53.                         columnQualifier = Bytes.toBytes(cxt.getConfiguration().get("conf.columnqualifier"));

  54.                         columnQualifier1 = Bytes.toBytes(cxt.getConfiguration().get("conf.columnqualifier1"));

  55.                 }

  56.                 @Override 

  57.                 protected void cleanup(Context cxt){

  58.                         try {

  59.                                 infoTable.put(list);

  60.                                 infoTable.flushCommits();

  61.                                 log.info("ParseListPutDriver cleanup ,current time :"+new Date());

  62.                         } catch (IOException e) {

  63.                                 log.error("infoTable flush commits error:\n"+e.getMessage());

  64.                         }

  65.                 }

  66.                 @Override 

  67.                 public void map(ImmutableBytesWritable row,Result columns,Context cxt){

  68.                         cxt.getCounter(Counters.ROWS).increment(1);

  69.                         String value =null ;

  70.                         try{

  71.                                 Put put = new Put(row.get());

  72.                                 for(KeyValue kv : columns.list()){

  73.                                         cxt.getCounter(Counters.COLS).increment(1);

  74.                                         value= Bytes.toStringBinary(kv.getValue());

  75.                                         if(equals(columnQualifier,kv.getQualifier())){  // 过滤column

  76.                                                 put.add(columnFamily,columnQualifier1,kv.getValue());

  77.                                                 list.add(put);

  78.                                         }

  79.                                 }

  80.                         }catch(Exception e){

  81.                                 log.info("Error:"+e.getMessage()+",Row:"+Bytes.toStringBinary(row.get())+

  82.                                                 ",Value:"+value);

  83.                                 cxt.getCounter(Counters.ERROR).increment(1);

  84.                         }

  85.                 }

  86.                 private boolean equals(byte[] a,byte[] b){

  87.                         String aStr= Bytes.toString(a);

  88.                         String bStr= Bytes.toString(b);

  89.                         if(aStr.equals(bStr)){

  90.                                 return true;

  91.                         }

  92.                         return false;

  93.                 }

  94.         }

  95.         public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

  96.                 String input ="testtable";

  97.                 byte[] columnFamily = Bytes.toBytes("fam");

  98.                 byte[] columnQualifier = Bytes.toBytes("data");

  99.                 byte[] columnQualifier1 = Bytes.toBytes("data2");

  100.                 Scan scan = new Scan ();

  101.                 scan.addColumn(columnFamily, columnQualifier);

  102.                 HadoopUtils.initialConf("hbase");

  103.                 Configuration conf = HadoopUtils.getConf();

  104.                 conf.set("conf.columnfamily", Bytes.toStringBinary(columnFamily));

  105.                 conf.set("conf.columnqualifier", Bytes.toStringBinary(columnQualifier));

  106.                 conf.set("conf.columnqualifier1", Bytes.toStringBinary(columnQualifier1));

  107.                 conf.set("conf.infotable", input);

  108.                 

  109.                 Job job = new Job(conf,"Parse data in "+input+",into tables");

  110.                 job.setJarByClass(ParseListPutDriver.class);

  111.                 TableMapReduceUtil.initTableMapperJob(input, scan, ParseMapper.class, 

  112.                                 ImmutableBytesWritable.class, Put.class,job);        

  113.                 job.setOutputFormatClass(NullOutputFormat.class);

  114.                 job.setNumReduceTasks(0);

  115.                 System.exit(job.waitForCompletion(true)?0:1);

  116.         }

  117. 

  118. }

复制代码

数据记录条数为：26632，可以看到下面图片中的时间记录对比：
1.jpg

2.png

由于结合了hbase，所以需要在hadoop_home/lib目录下面加些额外的包，其整个包如下(hbase1.0.jar为编译打包的MR程序)：
3.jpg






blog地址：http://blog.csdn.net/fansy1990

关注公众号，获取大数据、人工智能20套、区块链资源5阶段等资源，随时更新，获取最新技术资源
回复

使用道具 举报

自定义广告语

已有(8)人评论

跳转到指定楼层
正序浏览
释怀 发表于 2016-7-22 16:12:17
要是能多点注释对初学者来说就完美了
回复

使用道具 举报

天天哥们 发表于 2016-7-24 15:54:44
HBase高级应用实战视频教程—— http:// pan .baidu. com/s/1c2AzvOW 密码: n7vr
回复

使用道具 举报

George-zqq 发表于 2016-12-29 17:47:02
很好，学习学习
回复

使用道具 举报

sunbrillant 发表于 2017-9-1 22:57:35
之前用的都是MR读写，现在发现原来有这么多方式
回复

使用道具 举报

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
howtodown

版主

关注

  • 666

    主题

  • 1249

    帖子

  • 321

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000