HBase与MR的集成

r囧r小猫 2022-12-20 12:48 111阅读 0赞

MR   <--- 读取  HBase的数据
    
       MR   ---> 写入  HBase（使用场景：数据迁移）
    
       HBase ->  MR  -> HBase 读取和写入(使用场景：索引表的建立)

1.MR与HBase集成所需要的jar包

bin/hbase mapredcp

2.运行环境

export HADOOP_HOME=/opt/modules/hadoop
    export HBASE_HOME=/opt/modules/hbase
    HADOOP_CLASSPATH=`${ HBASE_HOME}/bin/hbase classpath` \
    ${HADOOP_HOME}/bin/hadoop jar ${HBASE_HOME}/lib/hbase-server-1.2.0-cdh5.9.3.jar

3.编写MR程序  
执行流程：将stu写入到stuMR表中

stu ->  mr  -> stuMR

package com.kfk.hbase;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.conf.Configured;
    import org.apache.hadoop.hbase.Cell;
    import org.apache.hadoop.hbase.CellUtil;
    import org.apache.hadoop.hbase.HBaseConfiguration;
    import org.apache.hadoop.hbase.client.Put;
    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.hbase.mapreduce.TableReducer;
    import org.apache.hadoop.hbase.util.Bytes;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    
    import org.apache.hadoop.util.Tool;
    import org.apache.hadoop.util.ToolRunner;
    
    import java.io.IOException;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/11/10 * @time : 7:07 下午 * * 需求分析，从数据表stu读取info到新表stuMR */
    
    public class HBaseMR extends Configured implements Tool { 
    
        /** * map * 一.Mapper class extends TableMapper<KEYOUT输出的Key的类型, VALUEOUT输出的Value的类型> * 原版的Mapper程序是有输入的KV类型，和输出的KV类型四个参数，源码：extends Mapper<ImmutableBytesWritable, * Result, KEYOUT, VALUEOUT> * Put类型为hbase中定义的类型，便于作为Reducer的输入类型,根据reducer输入类型可知 */
        public static class MyMapper extends TableMapper<Text, Put> { 
    
            private  Text mapOutputKey = new Text();
            public void map(ImmutableBytesWritable rowkey, Result value, Context context) throws InterruptedException, IOException { 
    
                // set outputRowKey
                mapOutputKey.set(Bytes.toString(rowkey.get()));
    
                // 通过rowKey创建put对象
                Put put = new Put(rowkey.get());
    
                // 迭代以获取cell数据
                for (Cell cell:value.rawCells()){ 
    
                    // 实际业务中不一定会是全表输出，可以选择性输出
                    if ("info".equals(Bytes.toString(CellUtil.cloneFamily(cell)))){ 
                        if ("username".equals(Bytes.toString(CellUtil.cloneQualifier(cell)))){ 
                            put.add(cell);
                        }
                        else if ("age".equals(Bytes.toString(CellUtil.cloneQualifier(cell)))){ 
                            put.add(cell);
                        }
                    }
                }
                context.write(mapOutputKey,put);
            }
        }
    
        /** * reduce * 二.Reducer calss extends TableReducer<KEYIN, VALUEIN, KEYOUT> * 输出key 类型为ImmutableBytesWritable 实现writeableComparable的字节数组 * 输出 value 类型为 Mutation 是 delete put increment append 的父类 */
        public static class MyReducer extends TableReducer<Text, Put, ImmutableBytesWritable> { 
    
            public void reduce(Text key, Iterable<Put> values, Context context) throws IOException, InterruptedException { 
    
                // 从得到的put中得到数据
                for (Put put:values){ 
                    // 往外写数据
                    context.write(null,put);
                }
            }
        }
    
        /** * run * @param args * @return * @throws IOException * @throws ClassNotFoundException * @throws InterruptedException */
        public int run(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
            // 1) get conf
            Configuration configuration = this.getConf();
    
            // 2) create job
            Job job = Job.getInstance(configuration,this.getClass().getSimpleName());
            job.setJarByClass(this.getClass());
    
            Scan scan = new Scan();
            scan.setCaching(500);
            scan.setCacheBlocks(false);
    
            // set input and set mapper
            TableMapReduceUtil.initTableMapperJob(
                    "stu",        // input table
                    scan,               // Scan instance to control CF and attribute selection
                    MyMapper.class,     // mapper class
                    Text.class,         // mapper output key
                    Put.class,  // mapper output value
                    job);
            // set reducer and output
            TableMapReduceUtil.initTableReducerJob(
                    "stuMR",        // output table
                    MyReducer.class,    // reducer class
                    job);
            
            // 4) commit，执行job
            boolean isSuccess = job.waitForCompletion(true);
            // 如果正常执行返回0，否则返回1
            return (isSuccess) ? 0 : 1;
        }
    
        public static void main(String[] args) { 
    
            Configuration configuration = HBaseConfiguration.create();
    
            try { 
    
                // 调用run方法
                int status = ToolRunner.run(configuration,new HBaseMR(),args);
                // 退出程序
                System.exit(status);
            } catch (IOException e) { 
                e.printStackTrace();
            } catch (ClassNotFoundException e) { 
                e.printStackTrace();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }

在本地运行的时候需要将core-site.xml文件的HDFS部分屏蔽掉

<configuration>
    
    
    
    
            <property>
            <name>hadoop.http.staticuser.user</name>
            <value>caizhengjie</value>
            </property>
    </configuration>

原数据：

hbase(main):003:0> scan 'stu'
    ROW                                                             COLUMN+CELL                                                                                                                                                                             
     001                                                            column=info:addres, timestamp=1604928754019, value=guangzhou                                                                                                                            
     001                                                            column=info:age, timestamp=1604928702206, value=20                                                                                                                                      
     001                                                            column=info:username, timestamp=1604928590845, value=alex                                                                                                                               
     002                                                            column=info:age, timestamp=1604955412618, value=30                                                                                                                                      
     002                                                            column=info:username, timestamp=1604955441414, value=ben                                                                                                                                
     003                                                            column=info:addres, timestamp=1604997979803, value=beijing                                                                                                                              
     003                                                            column=info:age, timestamp=1604997979803, value=32                                                                                                                                      
     003                                                            column=info:username, timestamp=1604997979803, value=jack                                                                                                                               
    3 row(s) in 0.0360 seconds

运行结果：

hbase(main):002:0> scan 'stuMR'
    ROW                                                             COLUMN+CELL                                                                                                                                                                             
     001                                                            column=info:age, timestamp=1604928702206, value=20                                                                                                                                      
     001                                                            column=info:username, timestamp=1604928590845, value=alex                                                                                                                               
     002                                                            column=info:age, timestamp=1604955412618, value=30                                                                                                                                      
     002                                                            column=info:username, timestamp=1604955441414, value=ben                                                                                                                                
     003                                                            column=info:age, timestamp=1604997979803, value=32                                                                                                                                      
     003                                                            column=info:username, timestamp=1604997979803, value=jack                                                                                                                               
    3 row(s) in 0.1780 seconds

4.打包发布运行命令：

export HADOOP_HOME=/opt/modules/hadoop
    export HBASE_HOME=/opt/modules/hbase
    HADOOP_CLASSPATH=`${ HBASE_HOME}/bin/hbase classpath` \
    ${HADOOP_HOME}/bin/hadoop jar /opt/jars/hbasemr.jar

由于我本地maven加载的是hbase-1.2.0版本，集群用的是hbase-1.2.0-cdh5.9.3版本，所以在运行jar包的时候会出现版本冲突：

Exception in thread "main" java.lang.NoSuchMethodError: org.apache.hadoop.hbase.client.Scan.setCaching(I)Lorg/apache/hadoop/hbase/client/Scan;
            at com.kfk.hbase.HBaseMR.run(HBaseMR.java:105)
            at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:70)
            at com.kfk.hbase.HBaseMR.main(HBaseMR.java:136)
            at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
            at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
            at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
            at java.lang.reflect.Method.invoke(Method.java:498)
            at org.apache.hadoop.util.RunJar.run(RunJar.java:221)
            at org.apache.hadoop.util.RunJar.main(RunJar.java:136)

我的解决方案是将hbase-client-1.2.0.jar移到hbase/lib目录下，将hbase-client-1.2.0-cdh5.9.3.jar改名，或者maven直接加载hbase的cdh版本，和集群的一致。

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！