數據在HDFS和HBASE之間互相傳遞的過程

　　對於簡單的結構化數據，我們在HDFS和HBASE上的傳遞可能只需要用框架即可完成，但是對於復雜的數據傳輸，特別是實際工作中，數據的收集整理並非簡單的結構，因此，我們需要對數據重新整理，並進行發送。這個過程就是依賴MapReduce，通過底層對數據的拆分和重組，達到我們要傳輸的結構要求。

下面我們開始進行一個簡單的小測試：

從HDFS 到HBASE

首先，我們在虛擬機本地創建一個臨時文件demo，簡單的結構化數據為name，age，sex。值可以適當添加幾行即可。然后通過命令

hdfs dfs -put demo 

 此時已經將文件存放到了HDFS上，然后在HBASE中，創建我們需要存儲數據的表，創建命令：

create 'tb_test','info'

 hbase 中表建立為表名tb_test,列族名info。

接下來我們編寫對應的代碼：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import java.io.IOException;

public class HdfstoHbase
{
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf=HBaseConfiguration.create();
        Job job=Job.getInstance(conf, "hdfs-to-hbase");
        job.setJarByClass(HdfstoHbase.class);
        job.setMapperClass(HdfsMap.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);
        TableMapReduceUtil.initTableReducerJob("tb_test",HbaseReduce.class,job );
        FileInputFormat.addInputPath(job,new Path("/user/bda/demo"));
        System.out.println(job.waitForCompletion(true)?0:1);


    }

    public static class HdfsMap extends Mapper<Object,Text,Text,NullWritable>{
        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            context.write(value,NullWritable.get() );
        }
    }
    public static class HbaseReduce extends TableReducer<Text,NullWritable,ImmutableBytesWritable>{
        private final byte[] cf="info".getBytes();
        private final byte[] name="name".getBytes();
        private final byte[] age="age".getBytes();
        private final byte[] sex="sex".getBytes();

        @Override
        protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
            String[] ss=key.toString().split(",");
            Put put=new Put(key.toString().getBytes());
            put.addColumn(cf,name ,ss[0].getBytes() );
            put.addColumn(cf,age ,ss[1].getBytes() );
            put.addColumn(cf,sex ,ss[2].getBytes() );
            context.write(null,put);
        }
    }
}

典型的MapReduce結構框架。然后打包，如果你打的是瘦包，那么，執行jar文件的過程你需要添加相應的依賴，或者是腳本提前寫好執行，下面直接貼出腳本執行命令：

HADOOP_CLASSPATH=`${HBASE_HOME}/bin/hbase classpath`:`${HBASE_HOME}/bin/hbase mapredcp` hadoop jar 打包的jar文件 文件的類名

執行后，在HBASE中，tb_test表即可查看到傳入的數據。

HBASE到HDFS

同樣的操作，這里只粘貼代碼部分，其余操作類似。

 

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class HbasetoHdfs {
    public static void main(String[] args) throws Exception {
        Configuration cfg = HBaseConfiguration.create();
        Job job = Job.getInstance(cfg,"hbase-to-hdfs");
        job.setJarByClass(HbasetoHdfs.class);
        Scan scan = new Scan();
        scan.setCaching(50);        // 1 is the default in Scan, which will be bad for MapReduce jobs
        scan.setCacheBlocks(false);
        TableMapReduceUtil.initTableMapperJob("tb_test",scan,HbaseMap.class,Text.class,NullWritable.class,job);
        job.setReducerClass(HdfsReduce.class);
        FileOutputFormat.setOutputPath(job,new Path("/user/bda/dsj3"));
        System.out.println(job.waitForCompletion(true)?0:1);


    }
    public static class HbaseMap extends TableMapper<Text,NullWritable> {
        private final byte[] cf="info".getBytes();
        private final byte[] name="name".getBytes();
        private final byte[] age="age".getBytes();
        private final byte[] sex="sex".getBytes();
        @Override
        protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
            String line="";
            String n = Bytes.toString(value.getValue(cf,name));
            String a = Bytes.toString(value.getValue(cf,age));
            String s = Bytes.toString(value.getValue(cf,sex));
            line = n+","+a+","+s;
            context.write(new Text(line),NullWritable.get());
        }
    }
    public static class HdfsReduce extends Reducer<Text,NullWritable,Text,NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
            context.write(key,NullWritable.get());
        }
    }

}