2.8.11 hadoop体系之离线计算-HBase数据库-HBase高级-HBase利用MapReduce

灰太狼 2022-12-26 09:26 127阅读 0赞

**目录**

1.写在前面

2.官方HBase-MapReduce配置

2.1 查看HBase 的MapReduce 任务的执行

2.2 环境变量导入（临时生效）

2.3 环境变量导入（永久生效）

3.简单案例：运行MapReduce任务

3.1 案例一：统计Student表中有多少行数据？

3.2 案例二：使用MR将本地数据导入到HBase

4.自定义HBase-MapReduce1

4.1 构建 FruitMapper 类，用于读取 fruit 表中的数据

4.2 构建 FruitReducer 类，用于将读取到的 fruit 表中的数据写入到 fruit\_mr表中

4.3 构建 FruitDriver extends Configured implements Tool 用于组装运行 Job任务

4.5 打包运行任务

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

# 1.写在前面 #

通过 HBase 的相关JavaAPI，我们可以实现伴随 HBase操作的MapReduce过程，比如使用 MapReduce 将数据从本地文件系统导入到 HBase 的表中，比如我们从 HBase 中读取一些原始数据后使用MapReduce 做数据分析。

# 2.官方HBase-MapReduce配置 #

## 2.1 查看HBase 的MapReduce 任务的执行 ##

bin/hbase mapredcp

## 2.2 环境变量导入（临时生效） ##

$ export HBASE_HOME=/opt/module/hbase 
    $ export HADOOP_HOME=/opt/module/hadoop-2.7.2 
    $ export HADOOP_CLASSPATH=`${HBASE_HOME}/bin/hbase mapredcp`

## 2.3 环境变量导入（永久生效） ##

在/etc/profile中配置

export HBASE_HOME=/opt/module/hbase 
    export HADOOP_HOME=/opt/module/hadoop-2.7.2

并在 hadoop-env.sh （cd /opt/software/hadoop/etc/hadoop/hadoop-env.sh）中配置：（注意：在 for 循环之后配）。

export HADOOP_CLASSPATH=$HADOOP_CLASSPATH:/opt/software/hbase/lib/*

注意：需要将这个文件分发到其他节点，并且重新启动hadoop，zk和hbase。

# 3.简单案例：运行MapReduce任务 #

启动方式：（stop对应着关闭）

（1）启动hdfs ** start-dfs.sh    **

（2）启动yarn   **start-yarn.sh**

（3）启动zk   **sh /root/bin/zkServer.sh start**

（4）启动hbase  ** start-hbase.sh**

关闭方式：

（1）关闭hbase   **stop-hbase.sh **

（2）关闭zk   ** sh /root/bin/zkServer.sh stop**

（3）关闭yarn   **stop-yarn.sh **

（4）关闭hdfs   **stop-dfs.sh**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

## 3.1 案例一：统计Student表中有多少行数据？ ##

官方案例

在hbase家目录下执行：

[root@h2node01 hbase-1.4.13]# yarn jar lib/hbase-server-1.4.13.jar rowcounter student

20/12/04 13:26:49 INFO mapreduce.Job:  map 0% reduce 0%
    20/12/04 13:26:59 INFO mapreduce.Job:  map 100% reduce 0%
    20/12/04 13:27:00 INFO mapreduce.Job: Job job_1607058845426_0001 completed successfully

![20201204133025185.png][]

## 3.2 案例二：使用MR将本地数据导入到HBase ##

1）在本地创建一个 tsv 格式的文件：fruit.tsv（中间分隔符用tab隔开）

1001    Apple    Red 
    1002    Pear    Yellow 
    1003    Pineapple    Yellow

2）创建Hbase 表  
Hbase(main):001:0> **create 'fruit','info'**

3）在HDFS 中创建 input\_fruit 文件夹并上传 fruit.tsv 文件

*  hdfs dfs -mkdir /input\_fruit/
 *  hdfs dfs -put fruit.tsv /input\_fruit/

4）执行 MapReduce 到HBase 的 fruit 表中

在hbase家目录下面执行：

yarn jar lib/hbase-server-1.4.13.jar importtsv -Dimporttsv.columns=HBASE_ROW_KEY,info:name,info:color fruit hdfs://h2node01:9000/input_fruit

5）使用 scan 命令查看导入后的结果  
Hbase(main):001:0> **scan 'fruit'**

![20201204164412837.png][]

# 4.自定义HBase-MapReduce1 #

目标：将Hbase的fruit 表中的一部分数据（name数据），通过MR迁入到 fruit\_mr 表中

分步实现：

## 4.1 构建 FruitMapper 类，用于读取 fruit 表中的数据 ##

package com.ucas.hbaseAndMr;
    
    import org.apache.hadoop.hbase.Cell;
    import org.apache.hadoop.hbase.CellUtil;
    import org.apache.hadoop.hbase.client.Put;
    import org.apache.hadoop.hbase.client.Result;
    import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
    import org.apache.hadoop.hbase.mapreduce.TableMapper;
    import org.apache.hadoop.hbase.util.Bytes;
    
    import java.io.IOException;
    
    /**
     * @author GONG
     * @version 1.0
     * @date 2020/12/4 20:29
     */
    public class FruitMapper extends TableMapper<ImmutableBytesWritable, Put> {
        @Override
        protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
            //构造put对象
            Put put = new Put(key.get());
    
            //遍历数据
            Cell[] cells = value.rawCells();
            for (Cell cell : cells) {
                if ("name".equals(Bytes.toString(CellUtil.cloneQualifier(cell)))) {
                    put.add(cell);
                }
            }
            //写出去
            context.write(key, put);
        }
    }

## 4.2 构建 FruitReducer 类，用于将读取到的 fruit 表中的数据写入到 fruit\_mr表中 ##

package com.ucas.hbaseAndMr;
    
    import org.apache.hadoop.hbase.client.Put;
    import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
    import org.apache.hadoop.hbase.mapreduce.TableReducer;
    import org.apache.hadoop.io.NullWritable;
    
    import java.io.IOException;
    
    /**
     * @author GONG
     * @version 1.0
     * @date 2020/12/4 20:30
     */
    public class FruitReducer extends TableReducer<ImmutableBytesWritable, Put, NullWritable> {
    
        @Override
        protected void reduce(ImmutableBytesWritable key, Iterable<Put> values, Context context) throws IOException, InterruptedException {
    
            //遍历写出
            for (Put value : values) {
                context.write(NullWritable.get(), value);
            }
        }
    }

## 4.3 构建 FruitDriver extends Configured implements Tool 用于组装运行 Job任务 ##

package com.ucas.hbaseAndMr;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.hbase.HBaseConfiguration;
    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;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.util.Tool;
    import org.apache.hadoop.util.ToolRunner;
    
    /**
     * @author GONG
     * @version 1.0
     * @date 2020/12/4 20:30
     */
    public class FruitDriver extends Configuration implements Tool {
    
        private Configuration configuration = null;
    
        public int run(String[] strings) throws Exception {
            //获取任务对象
            Job job = Job.getInstance(configuration);
            //指定Driver类
            job.setJarByClass(FruitDriver.class);
            //指定Mapper
            TableMapReduceUtil.initTableMapperJob("fruit", new Scan(), FruitMapper.class, ImmutableBytesWritable.class, Put.class, job);
            //指定Reducer
            TableMapReduceUtil.initTableReducerJob("fruit_mr", FruitReducer.class, job);
            //提交
            boolean b = job.waitForCompletion(true);
            return b ? 0 : 1;
        }
    
        public void setConf(Configuration conf) {
            this.configuration = conf;
        }
    
        public Configuration getConf() {
            return configuration;
        }
    
        //4．主函数中调用运行该 Job 任务
        public static void main(String[] args) throws Exception {
            Configuration configuration = HBaseConfiguration.create();
            int i = ToolRunner.run(configuration, new FruitDriver(), args);
        }
    }

## 4.5 打包运行任务 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1][]

创建接受数据的表

hbase(main):005:0>  create 'fruit_mr','info'

运行任务：

yarn jar hbaseMr-1.0-SNAPSHOT.jar com.ucas.hbaseAndMr.FruitDriver

查看导入的数据

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221120/89cac9581b624430b015f823f4ca6d33.png
[20201204133025185.png]: /images/20221120/8470f5fd0d174652aa8582b524f81f2a.png
[20201204164412837.png]: /images/20221120/6d1e84123edb4f2b8e43a8269316eb72.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1]: /images/20221120/d7e330ab49b14d6dbb9f75645b733f25.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2]: /images/20221120/0049f49453564985b3dfeb6393ce7c9f.png