HBase实现批量存取

package com.jiewen.hbase; import java.io.BufferedWriter; import java.io.File; import java.io.FileOutputStream; import java.io.FileWriter; import java.io.IOException; import java.io.PrintStream; import java.text.SimpleDateFormat; import java.util.Date; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.client.Durability; import org.apache.hadoop.hbase.client.Get; import org.apache.hadoop.hbase.client.HTable; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.hadoop.hbase.mapreduce.TableOutputFormat; import org.apache.hadoop.hbase.mapreduce.TableReducer; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Counter; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Mapper.Context; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; public class HbaseBatchOutput { public static void main(String[] args) throws Exception { final Configuration configuration = new Configuration(); // 设置zookeeper configuration.set("hbase.zookeeper.quorum", "localhost"); // 设置hbase表名称 configuration.set(TableOutputFormat.OUTPUT_TABLE, "zy2"); // 将该值改大，防止hbase超时退出 configuration.set("dfs.socket.timeout", "180000"); final Job job = new Job(configuration, "HBaseBatchImport"); job.setMapperClass(BatchImportMapper.class); job.setReducerClass(BatchImportReducer.class); // 设置map的输出，不设置reduce的输出类型 job.setMapOutputKeyClass(LongWritable.class); job.setMapOutputValueClass(Text.class); job.setInputFormatClass(TextInputFormat.class); // 不再设置输出路径，而是设置输出格式类型 job.setOutputFormatClass(TableOutputFormat.class); FileInputFormat.setInputPaths(job, "/user/root/input/result.txt"); job.waitForCompletion(true); } static class BatchImportMapper extends Mapper<LongWritable, Text, LongWritable, Text> { Text v2 = new Text(); protected void map(LongWritable key, Text value, Context context) throws java.io.IOException, InterruptedException { final String[] splited = value.toString().split("\n"); try{ // final LongWritable w = new // LongWritable(Long.parseLong(splited[0].trim())); // final Date date = new // Date(Long.parseLong(splited[0].trim())); // final String dateFormat = dateformat1.format(date); // final String dateFormat = a1.format(w); final String data = splited[0]; String rowKey = data; v2.set(rowKey); context.write(key, v2); } catch (NumberFormatException e) { final Counter counter = context.getCounter("BatchImport", "ErrorFormat"); counter.increment(1L); System.out.println("出错了" + splited[0] + " " + e.getMessage()); } }; } static class BatchImportReducer extends TableReducer<LongWritable, Text, NullWritable> { protected void reduce(LongWritable key, java.lang.Iterable<Text> values, Context context) throws java.io.IOException, InterruptedException { for (Text text : values) { final String[] splited = text.toString().split("\n"); // System.out.println(splited[0]); Configuration conf =HBaseConfiguration.create(); HTable table =new HTable(conf,"zy2") ; Get get =new Get(Bytes.toBytes(splited[0])); //get.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("data1")); Result result =table.get(get); byte[] val1 =result.getValue(Bytes.toBytes("cf"),Bytes.toBytes("data1")); //byte[] val2=result.getValue(Bytes.toBytes("cf"),Bytes.toBytes("data2")); //String val = Bytes.toString(val1)+" "+Bytes.toString(val2); String val = Bytes.toString(val1); // System.out.println(val); File f = new File("/home/zy/c.txt"); try { if(!f.exists()){ f.createNewFile(); } }catch(IOException e){ throw new RuntimeException("文件创建不成功!"); } FileOutputStream fos = new FileOutputStream(f, true); PrintStream ps = new PrintStream(fos); System.setOut(ps); System.out.println(val); /* FileWriter fw = null; File f = new File("/home/zy/3.txt"); try { if(!f.exists()){ f.createNewFile(); } fw = new FileWriter(f); BufferedWriter out = new BufferedWriter(fw); String[] s= val.split("\n"); for(int i=0;i<s.length;i++){ out.write(s[i]+"\r\n"); out.close(); } } catch (IOException e) { e.printStackTrace(); } */ } }; } }