您好,登錄后才能下訂單哦!
小編給大家分享一下MapReduce如何讀寫HBASE,相信大部分人都還不怎么了解,因此分享這篇文章給大家參考一下,希望大家閱讀完這篇文章后大有收獲,下面讓我們一起去了解一下吧!
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.StringTokenizer;
import java.util.Vector;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.commons.lang.Validate;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.eclipse.jdt.internal.codeassist.complete.CompletionOnArgumentName;
import sun.swing.MenuItemLayoutHelper.ColumnAlignment;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
public class MapHbase {
public static class TokenizerMapper_hbase extends TableMapper<Text, Text> {
public void map(ImmutableBytesWritable key, Result value, Context context)
throws IOException, InterruptedException {
List<KeyValue> kvs = value.list();
for (KeyValue val : kvs) {
System.out.println("column " + new String(val.getQualifier()) + " value " + new String(val.getValue()) + " key " + new String(val.getRow()));
String colname = new String(val.getQualifier());
String colvalue = new String(val.getValue());
context.write(new Text(val.getRow()), new Text(colname + "#" + colvalue));
}
}
}
public static class IntSumReducer_hbase extends TableReducer<Text, Text, ImmutableBytesWritable> {
public void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
Put put = new Put(Bytes.toBytes(key.toString()));
for (Text i : values) {
String val[] = i.toString().split("#");
if(val.length == 2){
String colname = val[0];
String colvalue = val[1];
put.add(Bytes.toBytes("cf"), Bytes.toBytes(colname), Bytes.toBytes(colvalue));
} else {
String colname = val[0];
put.add(Bytes.toBytes("cf"), Bytes.toBytes(colname), Bytes.toBytes(""));
}
}
context.write(null, put);
}
}
public static void main(String[] args) throws Exception {
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum",
"datanode01.isesol.com,datanode02.isesol.com,datanode03.isesol.com,datanode04.isesol.com,cmserver.isesol.com");
conf.set("hbase.zookeeper.property.clientPort", "2181");
Scan scan = new Scan();
scan.addFamily(Bytes.toBytes("cf"));
//scan.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("content"));
//scan.setMaxVersions();
Job job = new Job(conf, "t_ui_all");
job.setJarByClass(MapHbase.class);
// job.setMapperClass(TokenizerMapper_hbase.class);
// job.setMapOutputKeyClass(ImmutableBytesWritable.class);
// job.setMapOutputValueClass(Text.class);
TableMapReduceUtil.initTableMapperJob("t_ui_all", scan, TokenizerMapper_hbase.class, Text.class,
Text.class, job);
TableMapReduceUtil.initTableReducerJob("test2", IntSumReducer_hbase.class, job);
// FileInputFormat.addInputPath(job, new Path(args[0]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
以上是“MapReduce如何讀寫HBASE”這篇文章的所有內容,感謝各位的閱讀!相信大家都有了一定的了解,希望分享的內容對大家有所幫助,如果還想學習更多知識,歡迎關注億速云行業資訊頻道!
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。