HBase——MapReduce操作HBase

谁践踏了优雅 2022-05-26 07:16 340阅读 0赞

本节目标：

掌握编写MR操作HBase。

通过[官方实例][Link 1]，可以看到自定义map函数要继承TableMapper，自定义reduce函数要继承TableReducer。

![20180427114850712][]

![20180427114859141][]

![20180427114219486][]

测试实例：

/**
     * 从文本中读取内容，结果保存到HBase中
     */
    /**
     * map函数如果继承TableRMap，则从HBase中读取内容，在此依旧从HDFS中读取
     */
    class WCMap extends Mapper<LongWritable, Text, Text, IntWritable>{
    	/**
    	 * key: 偏移量
    	 * value:每行文本内容
    	 * context：输出内容上下文
    	 */
    	@Override
    	protected void map(LongWritable key, Text value,Context context)
    			throws IOException, InterruptedException {
    		IntWritable num = new IntWritable(1);
    		String str = value.toString();
    		String[] words = str.split(" ");
    		for (String word : words) {
    			context.write(new Text(word), num);
    		}
    	}	
    }
    /**
     * 继承TableReduce，结果保存到HBase表中
     * Text:输入的键类型
     * IntWritable：输入的值类型
     * ImmutableBytesWritable：输出的键类型
     */
    class WCReduce2 extends TableReducer<Text, IntWritable, ImmutableBytesWritable>{
    	@Override
    	protected void reduce(Text rowkey,Iterable<IntWritable> value,Context context)
    			throws IOException, InterruptedException {
    		int sum = 0;
    		for (IntWritable i : value) {
    			sum += i.get();
    		}
    		Put put = new Put(rowkey.toString().getBytes());
    		put.add("cf".getBytes(), "count".getBytes(), Bytes.toBytes(sum));
    		context.write(null, put);
    	}	
    }
    
    
    public class WCJob {
    	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    		//默认加载src目录下的配置文件（本地测试时，src下不能添加MR的配置文件）
    		Configuration conf = new Configuration();
    		
    		Job job = Job.getInstance(conf);//实例化job
    		
    		job.setJarByClass(WCJob.class);
    		
    		job.setMapperClass(WCMap.class);
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(IntWritable.class);
    		
    		Path inPath = new Path("/wc/input/wc.txt");//设置输入文件
    		FileInputFormat.addInputPath(job, inPath);
    		
    		//设置reduce
    		String targetTable = "wc";
    		TableMapReduceUtil.initTableReducerJob(targetTable, WCReduce2.class, job);
    		
    		boolean flag = job.waitForCompletion(true);
    		if (flag) {
    			System.out.println("Job Success!");
    		}
    	}
    }

本地测试会出错！

Exception in thread "main" java.lang.IllegalArgumentException: Pathname /D:/source\_maven/myhbase/lib/netty-all-4.0.50.Final.jar from

hdfs://master:8020/D:/source\_maven/myhbase/lib/netty-all-4.0.50.Final.jar is not a valid DFS filename.

at org.apache.hadoop.hdfs.DistributedFileSystem.getPathName(DistributedFileSystem.java:187)  
 at org.apache.hadoop.hdfs.DistributedFileSystem.access$000(DistributedFileSystem.java:101)  
 at org.apache.hadoop.hdfs.DistributedFileSystem$17.doCall(DistributedFileSystem.java:1068)  
 at org.apache.hadoop.hdfs.DistributedFileSystem$17.doCall(DistributedFileSystem.java:1064)  
 at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81)  
 at org.apache.hadoop.hdfs.DistributedFileSystem.getFileStatus(DistributedFileSystem.java:1064)  
 at org.apache.hadoop.mapreduce.filecache.ClientDistributedCacheManager.getFileStatus  
  
  
使用服务器环境测试：

1）src下添加4个配置文件

![2018042716452235][]

2）hadoop-env.sh 添加 HADOOP\_CLASSPATH=$HBASE\_HOME/lib/\*，否则会找不到hbase的jar包

[Link 1]: http://hbase.apache.org/book.html#mapreduce.example
[20180427114850712]: /images/20220526/9ee7ab1048d944ed8a24c82bb1d500ed.png
[20180427114859141]: /images/20220526/f7f2a6bf9040491eba3eb8892febff08.png
[20180427114219486]: /images/20220526/48a94a3de22c44d88a516ead397a44d9.png
[2018042716452235]: /images/20220526/dfb189c09aae4f60a15f580c2115b7c4.png