Hbase调用JavaAPI实现批量导入操作

1、Hbase调用JavaAPI实现批量导入操作将手机上网日志文件批量导入到Hbase中，操作步骤：1、将日志文件（请下载附件）上传到HDFS中，利用hadoop的操作命令上传：hadoopfs-putinput/2、创建Hbase表，通过Java操作Java代码packagecom.jiewen.hbase;importjava.io.IOException;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.hbase.HBaseConfiguration;importorg.apache.hadoop.hbas

2、e.HColumnDescriptor;importorg.apache.hadoop.hbase.HTableDescriptor;importorg.apache.hadoop.hbase.client.Get;importorg.apache.hadoop.hbase.client.HBaseAdmin;importorg.apache.hadoop.hbase.client.HTable;importorg.apache.hadoop.hbase.client.Put;importorg.apache.hadoop.hbase.client.Result;importorg.apach

3、e.hadoop.hbase.client.ResultScanner;importorg.apache.hadoop.hbase.client.Scan;importorg.apache.hadoop.hbase.util.Bytes;publicclassHbaseDemopublicstaticvoidmain(Stringargs)throwsIOExceptionStringtableName="wlan_log"StringcolumnFamily="cf"HbaseDemo.create(tableName,columnFamily);/H

4、baseDemo.put(tableName,"row1",columnFamily,"cl1","data");/HbaseDemo.get(tableName,"row1");/HbaseDemo.scan(tableName);/HbaseDemo.delete(tableName);/hbase操作必备privatestaticConfigurationgetConfiguration()Configurationconf=HBaseConfiguration.create();conf.set("

5、;hbase.rootdir","hdfs:/hadoop1:9000/hbase");/使用eclipse时必须添加这个，否则无法定位conf.set("hbase.zookeeper.quorum","hadoop1");returnconf;/创建一张表publicstaticvoidcreate(StringtableName,StringcolumnFamily)throwsIOExceptionHBaseAdminadmin=newHBaseAdmin(getConfiguration();if(admin.ta

6、bleExists(tableName)System.out.println("tableexists!");elseHTableDescriptortableDesc=newHTableDescriptor(tableName);tableDesc.addFamily(newHColumnDescriptor(columnFamily);admin.createTable(tableDesc);System.out.println("createtablesuccess!");/添加一条记录publicstaticvoidput(Stringtable

7、Name,Stringrow,StringcolumnFamily,Stringcolumn,Stringdata)throwsIOExceptionHTabletable=newHTable(getConfiguration(),tableName);Putp1=newPut(Bytes.toBytes(row);p1.add(Bytes.toBytes(columnFamily),Bytes.toBytes(column),Bytes.toBytes(data);table.put(p1);System.out.println("put'"+row+"

8、',"+columnFamily+":"+column+"','"+data+"'");/读取一条记录publicstaticvoidget(StringtableName,Stringrow)throwsIOExceptionHTabletable=newHTable(getConfiguration(),tableName);Getget=newGet(Bytes.toBytes(row);Resultresult=table.get(get);System.out.println(&qu

9、ot;Get:"+result);/显示所有数据publicstaticvoidscan(StringtableName)throwsIOExceptionHTabletable=newHTable(getConfiguration(),fortableName);Scanscan=newScan();ResultScannerscanner=table.getScanner(scan);(Resultresult:scanner)System.out.println("Scan:"+result);/删除表publicstaticvoiddelete(Strin

10、gtableName)throwsIOExceptionHBaseAdminadmin=newHBaseAdmin(getConfiguration();if(admin.tableExists(tableName)tryadmin.disableTable(tableName);admin.deleteTable(tableName);catch(IOExceptione)e.printStackTrace();System.out.println("Delete"+tableName+"失败");System.out.println("De

11、lete"+tableName+"成功");3、将日志文件导入Hbase表wlan_log中：Java代码importjava.text.SimpleDateFormat;importjava.util.Date;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.hbase.client.Put;importorg.apache.hadoop.hbase.mapreduce.TableOutputFormat;importorg.apache.hadoop.hbase.mapred

12、uce.TableReducer;importorg.apache.hadoop.hbase.util.Bytes;importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.NullWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Counter;importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.Mapper;importorg

13、.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.input.TextInputFormat;publicclassHbaseBatchImportpublicstaticvoidmain(Stringargs)throwsExceptionfinalConfigurationconfiguration=newConfiguration();/设置zookeeperconfiguration.set("hbase.zookeeper.quorum"

14、,"hadoop1");/设置hbase表名称configuration.set(TableOutputFormat.OUTPUT_TABLE,"wlan_log");/将该值改大，防止hbase超时退出configuration.set("dfs.socket.timeout","180000");finalJobjob=newJob(configuration,"HBaseBatchImport");job.setMapperClass(BatchImportMapper.class);jo

15、b.setReducerClass(BatchImportReducer.class);/设置map的输出，不设置reduce的输出类型job.setMapOutputKeyClass(LongWritable.class);job.setMapOutputValueClass(Text.class);job.setInputFormatClass(TextInputFormat.class);/不再设置输出路径，而是设置输出格式类型job.setOutputFormatClass(TableOutputFormat.class);FileInputFormat.setInputPaths(j

16、ob,"hdfs:/hadoop1:9000/input");job.waitForCompletion(true);staticclassBatchImportMapperextendsMapper<LongWritable,Text,LongWritable,Text>SimpleDateFormatdateformat1=newSimpleDateFormat("yyyyMMddHHmmss");Textv2=newText();protectedvoidmap(LongWritablekey,Textvalue,Con

17、textcontext)throwsjava.io.IOException,InterruptedExceptionfinalStringsplited=value.toString().split("t");tryfinalDatedate=newDate(Long.parseLong(splited0.trim();finalStringdateFormat=dateformat1.format(date);StringrowKey=splited1+":"+dateFormat;v2.set(rowKey+"t"+value.t

18、oString();context.write(key,v2);catch(NumberFormatExceptione)finalCountercounter=context.getCounter("BatchImport","ErrorFormat");counter.increment(1L);System.out.println("出错了"+splited0+""+e.getMessage();staticclassBatchImportReducerextendsTableReducer<LongWritable,Text,NullWritable>protectedvoidreduce(LongWritablekey,java.lang.Iterable<Text>values,Contextcontext)throwsjava.io.IOException,Interrup