HBase和HDFS數(shù)據(jù)互導(dǎo)程序
下面說說JAVA API提供的這些類的功能和他們之間有什么樣的聯(lián)系。
公司主營業(yè)務(wù)：做網(wǎng)站、成都做網(wǎng)站、移動網(wǎng)站開發(fā)等業(yè)務(wù)。幫助企業(yè)客戶真正實(shí)現(xiàn)互聯(lián)網(wǎng)宣傳，提高企業(yè)的競爭能力。創(chuàng)新互聯(lián)公司是一支青春激揚(yáng)、勤奮敬業(yè)、活力青春激揚(yáng)、勤奮敬業(yè)、活力澎湃、和諧高效的團(tuán)隊(duì)。公司秉承以“開放、自由、嚴(yán)謹(jǐn)、自律”為核心的企業(yè)文化，感謝他們對我們的高要求，感謝他們從不同領(lǐng)域給我們帶來的挑戰(zhàn)，讓我們激情的團(tuán)隊(duì)有機(jī)會用頭腦與智慧不斷的給客戶帶來驚喜。創(chuàng)新互聯(lián)公司推出紫陽免費(fèi)做網(wǎng)站回饋大家。
1.HBaseConfiguration
關(guān)系：org.apache.hadoop.hbase.HBaseConfiguration
作用：通過此類可以對HBase進(jìn)行配置
用法實(shí)例： Configuration config = HBaseConfiguration.create();
說明： HBaseConfiguration.create()默認(rèn)會從classpath中查找 hbase-site.xml中的配置信息，初始化 Configuration。
2.HBaseAdmin 類
關(guān)系：org.apache.hadoop.hbase.client.HBaseAdmin
作用：提供接口關(guān)系HBase數(shù)據(jù)庫中的表信息
用法：HBaseAdmin admin = new HBaseAdmin(config);
3.Descriptor類
關(guān)系：org.apache.hadoop.hbase.HTableDescriptor
作用：HTableDescriptor類包含了表的名字以及表的列族信息
用法：HTableDescriptor htd =new HTableDescriptor(tablename);
構(gòu)造一個表描述符指定TableName對象。
Htd.addFamily(new HColumnDescriptor(“myFamily”));
將列家族給定的描述符
4.HTable
關(guān)系：org.apache.hadoop.hbase.client.HTable
作用：HTable和 HBase的表通信
用法：HTable tab = new HTable(config,Bytes.toBytes(tablename));
ResultScanner sc = tab.getScanner(Bytes.toBytes(“familyName”));
說明：獲取表內(nèi)列族 familyNme的所有數(shù)據(jù)。
5.Put
關(guān)系：org.apache.hadoop.hbase.client.Put
作用：獲取單個行的數(shù)據(jù)
用法：HTable table = new HTable(config,Bytes.toBytes(tablename));
Put put = new Put(row);
p.add(family,qualifier,value);
說明：向表 tablename添加 “family,qualifier,value”指定的值。
6.Get
關(guān)系：org.apache.hadoop.hbase.client.Get
作用：獲取單個行的數(shù)據(jù)
用法：HTable table = new HTable(config,Bytes.toBytes(tablename));
Get get = new Get(Bytes.toBytes(row));
Result result = table.get(get);
說明：獲取 tablename表中 row行的對應(yīng)數(shù)據(jù)
7.ResultScanner
關(guān)系：Interface
作用：獲取值的接口
用法：ResultScanner scanner = table.getScanner(Bytes.toBytes(family));
For(Result rowResult : scanner){
Bytes[] str = rowResult.getValue(family,column);
}
說明：循環(huán)獲取行中列值。
例1 HBase之讀取HDFS數(shù)據(jù)寫入HBase
package org.hadoop.hbase;
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
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.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
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 org.apache.hadoop.util.GenericOptionsParser;
public class WordCountHbaseWriter {
public static class WordCountHbaseMapper extends
Mapper {
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
public void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
StringTokenizer itr = new StringTokenizer(value.toString());
while (itr.hasMoreTokens()) {
word.set(itr.nextToken());
context.write(word, one);// 輸出為
}
}
}
public static class WordCountHbaseReducer extends
TableReducer {
public void reduce(Text key, Iterable values,
Context context) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {// 遍歷求和
sum += val.get();
}
Put put = new Put(key.getBytes());//put實(shí)例化，每一個詞存一行
//列族為content,列修飾符為count，列值為數(shù)目
put.add(Bytes.toBytes("content"), Bytes.toBytes("count"), Bytes.toBytes(String.valueOf(sum)));
context.write(new ImmutableBytesWritable(key.getBytes()), put);// 輸出求和后的
}
}
public static void main(String[] args){
String tablename = "wordcount";
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum", "192.168.1.139");
conf.set("hbase.zookeeper.property.clientPort", "2191");
HBaseAdmin admin = null;
try {
admin = new HBaseAdmin(conf);
if(admin.tableExists(tablename)){
System.out.println("table exists!recreating.......");
admin.disableTable(tablename);
admin.deleteTable(tablename);
}
HTableDescriptor htd = new HTableDescriptor(tablename);
HColumnDescriptor tcd = new HColumnDescriptor("content");
htd.addFamily(tcd);//創(chuàng)建列族
admin.createTable(htd);//創(chuàng)建表
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 1) {
System.err.println("Usage: WordCountHbaseWriter ");
System.exit(2);
}
Job job = new Job(conf, "WordCountHbaseWriter");
job.setNumReduceTasks(2);
job.setJarByClass(WordCountHbaseWriter.class);
//使用WordCountHbaseMapper類完成Map過程；
job.setMapperClass(WordCountHbaseMapper.class);
TableMapReduceUtil.initTableReducerJob(tablename, WordCountHbaseReducer.class, job);
//設(shè)置任務(wù)數(shù)據(jù)的輸入路徑；
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
//設(shè)置了Map過程的輸出類型，其中設(shè)置key的輸出類型為Text；
job.setOutputKeyClass(Text.class);
//設(shè)置了Map過程的輸出類型，其中設(shè)置value的輸出類型為IntWritable；
job.setOutputValueClass(IntWritable.class);
//調(diào)用job.waitForCompletion(true) 執(zhí)行任務(wù)，執(zhí)行成功后退出；
System.exit(job.waitForCompletion(true) ? 0 : 1);
} catch (Exception e) {
e.printStackTrace();
} finally{
if(admin!=null)
try {
admin.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
例2 HBase之讀取HBase數(shù)據(jù)寫入HDFS
package org.hadoop.hbase;
import java.io.IOException;
import java.util.Map.Entry;
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.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
public class WordCountHbaseReader {
public static class WordCountHbaseReaderMapper extends
TableMapper{
@Override
protected void map(ImmutableBytesWritable key,Result value,Context context)
throws IOException, InterruptedException {
StringBuffer sb = new StringBuffer("");
for(Entry entry:value.getFamilyMap("content".getBytes()).entrySet()){
String str = new String(entry.getValue());
//將字節(jié)數(shù)組轉(zhuǎn)換為String類型
if(str != null){
sb.append(new String(entry.getKey()));
sb.append(":");
sb.append(str);
}
context.write(new Text(key.get()), new Text(new String(sb)));
}
}
}
public static class WordCountHbaseReaderReduce extends Reducer{
private Text result = new Text();
@Override
protected void reduce(Text key, Iterable values,Context context)
throws IOException, InterruptedException {
for(Text val:values){
result.set(val);
context.write(key, result);
}
}
}
public static void main(String[] args) throws Exception {
String tablename = "wordcount";
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum", "192.168.1.139");
conf.set("hbase.zookeeper.property.clientPort", "2191");
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 1) {
System.err.println("Usage: WordCountHbaseReader ");
System.exit(2);
}
Job job = new Job(conf, "WordCountHbaseReader");
job.setJarByClass(WordCountHbaseReader.class);
//設(shè)置任務(wù)數(shù)據(jù)的輸出路徑；
FileOutputFormat.setOutputPath(job, new Path(otherArgs[0]));
job.setReducerClass(WordCountHbaseReaderReduce.class);
Scan scan = new Scan();
TableMapReduceUtil.initTableMapperJob(tablename,scan,WordCountHbaseReaderMapper.class, Text.class, Text.class, job);
//調(diào)用job.waitForCompletion(true) 執(zhí)行任務(wù)，執(zhí)行成功后退出；
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
} 程序中用到hadoop的相關(guān)JAR包（如下圖）及hbase所有jar包
如果上面的API還不能滿足你的要求，可以到下面這個網(wǎng)站里面Hbase全部API介紹
http://www.yiibai.com/hbase/
當(dāng)前標(biāo)題：HBase和HDFS數(shù)據(jù)互導(dǎo)程序
本文URL：http://m.jiaotiyi.com/article/iiohhi.html