建表

package hbase_hdfs;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.util.Bytes;



public class CreatTable {


	public static Configuration configuration;
	public static Admin admin;
	public static Connection connection;
	public static void main(String[] args) throws IOException {
		init();
		creatTable("WordCount",new String[] {"word"});
		insertData("WordCount", "1", "word", "1", "hadoop,hbase,zookeeper");
		insertData("WordCount", "2", "word", "2", "hello,hbase,world");
		insertData("WordCount", "3", "word", "3", "hello,hadoop,world");
		insertData("WordCount", "4", "word", "4", "hello,zookeeper,world");
		close();
	}
	
    //创建连接
    
	public static void init() throws IOException
	{
		configuration = HBaseConfiguration.create();
		configuration.set("hbase.rootdir", "hdfs://localhost:9000/hbase");
		connection = ConnectionFactory.createConnection(configuration);
		admin = connection.getAdmin();
		System.out.println("连接成功");
	}

    //关闭连接
    
	public static void close()
	{
		if(admin!=null)
		{
			admin.close();
			
		}
		if(connection!=null)
		{
			try {
				connection.close();
			} catch (IOException e) {

				e.printStackTrace();
			}
		}
	}
	public static void creatTable(String myTableName,String[] colFamily) throws IOException
	{
		TableName tableName = TableName.valueOf(myTableName);
        
		//判断是否已经存在
        
		if(admin.tableExists(tableName))
		{
			System.out.println("该表已存在，正在删除");
			admin. disableTableAsync(TableName.valueOf(myTableName));
        	admin.deleteTable(TableName.valueOf(myTableName));
					
		}
		TableDescriptorBuilder tableDescriptor = TableDescriptorBuilder.newBuilder(tableName);
		for(String str:colFamily)
		{
			ColumnFamilyDescriptor family = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(str)).build();
			tableDescriptor.setColumnFamily(family);
			admin.createTable(tableDescriptor.build());
			
		}
	}
	public static void insertData(String tableName,String rowKey,String colFamily,String col,String val) throws IOException 
	{
		Table table = connection.getTable(TableName.valueOf(tableName));
		Put put = new Put(rowKey.getBytes());
		put.addColumn(colFamily.getBytes(),col.getBytes(), val.getBytes());
		table.put(put);
		table.close();
		System.out.println("插入成功");
	}
}

MapReduce从hbase中读数据

package hbase_hdfs_;

import org.apache.hadoop.conf.Configuration;
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.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;
public class WordCount {
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		 Configuration conf = HBaseConfiguration.create();
	        conf.set("hbase.zookeeper.quorum", "hadoop01");
	        String tableName = "WordCount";
	        Job job = new Job(conf,"WordCountHbase");
	        job.setJarByClass(WordCount.class);
	        Scan scan =new Scan();
	        TableMapReduceUtil.initTableMapperJob(tableName,scan, MyMapper.class, Text.class, IntWritable.class, job);
	        job.setReducerClass(WordReducer.class);
	        job.setOutputKeyClass(WritableComparable.class);
	        job.setOutputValueClass(Writable.class);
	        FileOutputFormat.setOutputPath(job, new Path( "hdfs://hadoop01:9000/HbaseOutput"));
	        System.exit(job.waitForCompletion(true)?0:1);

	}
	private static class MyMapper extends TableMapper<Text,IntWritable>
	{
		private final static IntWritable one = new IntWritable(1);
		private Text word = new Text();
		@Override
		protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException 
		{
            for(Cell cell:value.rawCells())
        	{
				word.set(CellUtil.cloneValue(cell));
				String[] lines = word.toString().split(",");
				for(String words:lines)
				{
					context.write(new Text(words), one);
				}
			}
		}
		public static class WordReducer extends Reducer<Text,IntWritable,Text,IntWritable> 
		{
	        @Override
	        public void reduce(Text key,Iterable<IntWritable> values,Context context) throws IOException, InterruptedException 
	        {

	            int sum =0;
	            for(IntWritable val:values){
	                sum++;
	            }
	            context.write(key, new IntWritable(sum));
	        }
	    }
		
	}
}

结果：