HBase本身提供了很多種資料匯入的方式，通常有兩種常用方式：

1. 使用HBase提供的TableOutputFormat，原理是通過一個Mapreduce作業將資料匯入HBase
2. 另一種方式就是使用HBase原生Client API

本文就是示範如何通過MapReduce作業從一個檔案讀取資料並寫入到HBase中。

首先啟動Hadoop與HBase，然後建立一個空表，用於後面匯入資料：

hbase(main):006:0> create 'mytable','cf'
0 row(s) in 10.8310 seconds

=> Hbase::Table - mytable
hbase(main):007
 
:0> list
TABLE                                                                                                   
mytable                                                                                                 
1 row(s) in 0.1220 seconds

=> ["mytable"]
hbase(main):008:0> scan 'mytable'
ROW                         COLUMN+CELL                                                                 
0
 
 row(s) in 0.2130 seconds

一、示例程式

下面的示例程式通過TableOutputFormat將HDFS上具有一定格式的文字資料匯入到HBase中。

首先建立MapReduce作業，目錄結構如下：

Hdfs2HBase/
├── classes
└── src
    ├── Hdfs2HBase.java
    ├── Hdfs2HBaseMapper.java
    └── Hdfs2HBaseReducer.java

Hdfs2HBaseMapper.java

package com.lisong.hdfs2hbase;

import java.io.IOException;

import
 
 org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class Hdfs2HBaseMapper extends Mapper<LongWritable, Text, Text, Text> {
        public void map(LongWritable key, Text line, Context context) throws IOException,InterruptedException {
                String lineStr = line.toString();
                int index = lineStr.indexOf(":");
                String rowkey = lineStr.substring(0, index);
                String left = lineStr.substring(index+1);
                context.write(new Text(rowkey), new Text(left));
        }
}

Hdfs2HBaseReducer.java

package com.lisong.hdfs2hbase;

import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

public class Hdfs2HBaseReducer extends Reducer<Text, Text, ImmutableBytesWritable, Put> {
        public void reduce(Text rowkey, Iterable<Text> value, Context context) throws IOException,InterruptedException {
                String k = rowkey.toString();
                for(Text val : value) {
                        Put put = new Put(k.getBytes());
                        String[] strs = val.toString().split(":");
                        String family = strs[0];
                        String qualifier = strs[1];
                        String v = strs[2];
                        put.add(family.getBytes(), qualifier.getBytes(), v.getBytes());
                        context.write(new ImmutableBytesWritable(k.getBytes()), put);
                }
        }
}

Hdfs2HBase.java

package com.lisong.hdfs2hbase;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class Hdfs2HBase {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if(otherArgs.length != 2) {
            System.err.println("Usage: wordcount <infile> <table>");
            System.exit(2);
        }

        Job job = new Job(conf, "hdfs2hbase");
        job.setJarByClass(Hdfs2HBase.class);
        job.setMapperClass(Hdfs2HBaseMapper.class);
        job.setReducerClass(Hdfs2HBaseReducer.class);

        job.setOutputKeyClass(ImmutableBytesWritable.class);
        job.setOutputValueClass(Put.class);

        job.setOutputFormatClass(TableOutputFormat.class);

        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, otherArgs[1]);

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

配置javac編譯依賴環境：

$HADOOP_HOME/share/hadoop/common/hadoop-common-2.4.1.jar
$HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-client-core-2.4.1.jar
$HADOOP_HOME/share/hadoop/common/lib/commons-cli-1.2.jar

這裡要操作HBase，故除了上面三個jar包，還需要$HBASE_HOME/lib目錄下的jar包。為了方便，我們在/etc/profile的CLASSPATH裡包含所有的依賴包：

TEMP=`ls /home/hadoop/hbase/lib/*.jar`
HBASE_JARS=`echo $TEMP | sed 's/ /:/g'`
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:/home/hadoop/hadoop/share/hadoop/common/hadoop-common-2.6.0.jar:/home/hadoop/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-client-core-2.6.0.jar:/home/hadoop/hadoop/share/hadoop/common/lib/commons-cli-1.2.jar:$HBASE_JARS

編譯

$ javac -d classes/ src/*.java

打包

$ jar -cvf hdfs2hbase.jar classes

執行

建立一個data.txt檔案，內容如下（列族是建表時建立的列族cf）：

r1:cf:c1:value1 
r2:cf:c2:value2 
r3:cf:c3:value3

將檔案複製到hdfs上：

$ hadoop/bin/hadoop fs -put data.txt /hbase

執行MapReduce作業：

$ hadoop/bin/hadoop jar Hdfs2HBase/hdfs2hbase.jar com.lisong.hdfs2hbase.Hdfs2HBase /hbase/data.txt mytable

報錯NoClassDefFoundError找不到類定義：

Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/hadoop/hbase/io/ImmutableBytesWritable
    at com.lisong.hdfs2hbase.Hdfs2HBase.main(Hdfs2HBase.java:30)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    ...
    at org.apache.hadoop.util.RunJar.run(RunJar.java:221)
    at org.apache.hadoop.util.RunJar.main(RunJar.java:136)
Caused by: java.lang.ClassNotFoundException: org.apache.hadoop.hbase.io.ImmutableBytesWritable
    at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
    ... 7 more

原因是我沒有把HBase的jar包加到hadoop-env.sh中。

TEMP=`ls /home/hadoop/hbase/lib/*.jar`
HBASE_JARS=`echo $TEMP | sed 's/ /:/g'`
HADOOP_CLASSPATH=$HBASE_JARS

再次執行發現又報了Unable to initialize MapOutputCollector的錯誤：

15/08/10 08:55:44 WARN mapred.MapTask: Unable to initialize MapOutputCollector org.apache.hadoop.mapred.MapTask$MapOutputBuffer
java.lang.NullPointerException
    at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.init(MapTask.java:1008)
    at org.apache.hadoop.mapred.MapTask.createSortingCollector(MapTask.java:401)
    ...
    at java.lang.Thread.run(Thread.java:745)
15/08/10 08:55:44 INFO mapred.LocalJobRunner: map task executor complete.
15/08/10 08:55:44 WARN mapred.LocalJobRunner: job_local2138114942_0001
java.lang.Exception: java.io.IOException: Unable to initialize any output collector
    at org.apache.hadoop.mapred.LocalJobRunner$Job.runTasks(LocalJobRunner.java:462)
    at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:522)
Caused by: java.io.IOException: Unable to initialize any output collector
    at org.apache.hadoop.mapred.MapTask.createSortingCollector(MapTask.java:412)
    ...
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
15/08/10 08:55:44 INFO mapreduce.Job: Job job_local2138114942_0001 failed with state FAILED due to: NA
15/08/10 08:55:45 INFO mapreduce.Job: Counters: 0

原因是我沒有指明Map輸出的Key/Value型別，在Hdfs2HBase.java中新增以下兩句：

job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);     

如果沒有專門定義Mapper輸出型別的話，job.setOutputKeyClass和job.setOutputValueClass設定的是Mapper和Reducer兩個的輸出型別。

job.setOutputKeyClass(ImmutableBytesWritable.class);
job.setOutputValueClass(Put.class);

而Hdfs2HBaseMapper輸出型別是Text/Text，所以這裡需要單獨指定。

修改Hdfs2HBase.java

package com.lisong.hdfs2hbase;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class Hdfs2HBase {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if(otherArgs.length != 2) {
            System.err.println("Usage: wordcount <infile> <table>");
            System.exit(2);
        }

        Job job = new Job(conf, "hdfs2hbase");
        job.setJarByClass(Hdfs2HBase.class);
        job.setMapperClass(Hdfs2HBaseMapper.class);
        job.setReducerClass(Hdfs2HBaseReducer.class);

        job.setMapOutputKeyClass(Text.class);    // +
        job.setMapOutputValueClass(Text.class);  // +

        job.setOutputKeyClass(ImmutableBytesWritable.class);
        job.setOutputValueClass(Put.class);

        job.setOutputFormatClass(TableOutputFormat.class);

        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, otherArgs[1]);

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

再次編譯、打包，然後執行成功！

查詢HBase表，驗證資料是否已匯入：

hbase(main):001:0> scan 'mytable'
ROW                         COLUMN+CELL                                                                 
 r1                         column=cf:c1, timestamp=1439223857492, value=value1                         
 r2                         column=cf:c2, timestamp=1439223857492, value=value2                         
 r3                         column=cf:c3, timestamp=1439223857492, value=value3                         
3 row(s) in 1.3820 seconds

可以看到，資料匯入成功！

由於需要頻繁的與儲存資料的RegionServer通訊，佔用資源較大，一次性入庫大量資料時，TableOutputFormat效率並不好。

二、拓展-TableReducer

我們可以將Hdfs2HBaseReducer.java程式碼改成下面這樣，作用是一樣的：

package com.lisong.hdfs2hbase;

import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

public class Hdfs2HBaseReducer extends TableReducer<Text, Text, ImmutableBytesWritable> {
    public void reduce(Text rowkey, Iterable<Text> value, Context context) throws IOException,InterruptedException {
        String k = rowkey.toString();
        for(Text val : value) {
            Put put = new Put(k.getBytes());
            String[] strs = val.toString().split(":");
            String family = strs[0];
            String qualifier = strs[1];
            String v = strs[2];
            put.add(family.getBytes(), qualifier.getBytes(), v.getBytes());
            context.write(new ImmutableBytesWritable(k.getBytes()), put);
        }
    }
}

這裡直接繼承了TableReducer，TableReducer是部分特例化的Reducer，它只有三個型別引數：輸入Key/Value是對應Mapper的輸出，輸出Key可以是任意的型別，但是輸出Value必須是一個Put或Delete例項。

編譯打包執行，結果與前面的一樣！