实现一个工具类，可以把HBase任意表的任意多的列导出到任意指定的HDFS中

package hbase;

import java.io.IOException;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
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.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class HBase2HdfsUtils {
/**
* args[0]  表名
* args[1]	列族、列名称列表，格式---列族:列
* args[2]  输出路径
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception {
//获取Hbase的配置信息，从resources目录下的hbase-site.xml文件中获取配置信息
Configuration conf = HBaseConfiguration.create();
//设置列族、列名称信息列表参数,格式--列族:列
conf.set("FamilyColumnsList", args[1]);

//申明一个客户端
Job job = Job.getInstance(conf, HBase2HdfsUtils.class.getSimpleName());
//打成jar包执行需要指定类名
job.setJarByClass(HBase2HdfsUtils.class);

//指定HBase中需要导出表的信息，即map的输入
Scan scan = new Scan();
TableMapReduceUtil.initTableMapperJob(args[0], scan, MyMapper.class, Text.class,  Text.class, job);

//设置输入的配置信息：key、value的类型
job.setMapperClass(MyMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);

//从hbase导出数据到hdfs不需要reduce，所以设置reduce的任务数为0
job.setNumReduceTasks(0);

//设置输出的配置信息：key、value的类型以及输出路径
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
job.setOutputFormatClass(TextOutputFormat.class);

//如果输出目录存在，则删除输出目录
Path path = new Path(args[2]);
FileSystem fs = FileSystem.get(new URI(args[2]), new Configuration());
if(fs.exists(path)){
fs.delete(path, true);
}

FileOutputFormat.setOutputPath(job, new Path(args[2]));

job.waitForCompletion(true);
}
static class MyMapper extends  TableMapper<Text, Text>{
Text k2 = new Text();
Text v2 = new Text();
@Override
protected void map(
ImmutableBytesWritable key,
Result value,
Mapper<ImmutableBytesWritable, Result, Text, Text>.Context context)
throws IOException, InterruptedException {
k2.set("");

String v2Text = "";
String familyColumnsList = context.getConfiguration().get("FamilyColumnsList");
String[] splited = familyColumnsList.split(",");

String title = "";  //标题
for (String split : splited) {
String[] column = split.split(":");
//根据列族、列获取值
Cell cell = value.getColumnLatestCell(column[0].getBytes(), column[1].getBytes());
//判断据列族、列获取到的cell不为空，否则会报空指针错误
if(cell!=null){
title += new String(CellUtil.cloneQualifier(cell)) + ":" +  new String(CellUtil.cloneQualifier(cell)) +  "\t" ;
v2Text += new String(CellUtil.cloneValue(cell)) +  "\t" ;
}

}

v2.set(title + "\n" + v2Text);
context.write(k2, v2);
}
}

}