专栏首页不温卜火HBase快速入门系列(7) | 官方HBase-MapReduce与自定义

HBase快速入门系列(7) | 官方HBase-MapReduce与自定义

1. 官方HBase-MapReduce

1.查看HBase的MapReduce任务的执行

[bigdata@hadoop002 hbase]$ bin/hbase mapredcp

上图标记处为所需jar包

2. 环境变量的导入

  • 1. 执行环境变量的导入(临时生效,在命令行执行下述操作)
$ export HBASE_HOME=/opt/module/hbase
$ export HADOOP_HOME=/opt/module/hadoop-2.7.2
$ export HADOOP_CLASSPATH=`${HBASE_HOME}/bin/hbase mapredcp`

// 也可以直接这样
[bigdata@hadoop002 hbase]$ export HADOOP_CLASSPATH=`/opt/module/hbase/bin/hbase mapredcp`

配置完成后查看是否成功

  • 2. 永久生效:在/etc/profile配置
export HBASE_HOME=/opt/module/hbase
export HADOOP_HOME=/opt/module/hadoop-2.7.2

在hadoop-env.sh中配置:(注意:在for循环之后配)

export HADOOP_CLASSPATH=$HADOOP_CLASSPATH:/opt/module/hbase/lib/*
  • 3. 运行官方的MapReduce任务

– 案例一:统计Student表中有多少行数据

[bigdata@hadoop002 hbase]$ /opt/module/hadoop-2.7.2/bin/yarn jar lib/hbase-server-1.3.1.jar rowcounter student

– 案例二:使用MapReduce将HDFS导入到HBase

  • 1.在本地创建一个tsv格式的文件:fruit.tsv
[bigdata@hadoop002 datas]$ vim fruit.tsv

1001	Apple	Red
1002	Pear	Yellow
1003	Pineapple	Yellow
  • 2.创建HBase表
hbase(main):001:0> create 'fruit','info'
  • 3.在HDFS中创建input_fruit文件夹并上传fruit.tsv文件
[bigdata@hadoop002 datas]$ hadoop fs -mkdir /input_fruit/
[bigdata@hadoop002 datas]$ hadoop fs -put fruit.tsv /input_fruit/
  • 4.执行MapReduce到HBase的fruit表中
[bigdata@hadoop002 hbase]$ /opt/module/hadoop-2.7.2/bin/yarn jar lib/hbase-server-1.3.1.jar importtsv \
-Dimporttsv.columns=HBASE_ROW_KEY,info:name,info:color fruit \
hdfs://hadoop002:9000/input_fruit
  • 5.使用scan命令查看导入后的结果
hbase(main):001:0> scan ‘fruit’

经过测试证明是没问题的

2. 自定义HBase-MapReduce1

目标:将fruit表中的一部分数据,通过MR迁入到fruit_mr表中。

  • 1.构建ReadMapper类,用于读取fruit表中的数据
package com.buwenbuhuo.hbase.mr;

import java.io.IOException;
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;


/**
 * @author 卜温不火
 * @create 2020-05-12 19:32
 * com.buwenbuhuo.hbase.mr - the name of the target package where the new class or interface will be created.
 * hbase0512 - the name of the current project.
 */
public class ReadMapper extends TableMapper<ImmutableBytesWritable,Put> {

    @Override
    protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
        Put put = new Put(key.copyBytes());

        for (Cell cell : value.rawCells()) {
            if ("name".equals(Bytes.toString(CellUtil.cloneQualifier(cell)))){

                put.add(cell);

            }
        }
        context.write(key,put);
    }
}
  • 2. 构建WriteReducer类,用于将读取到的fruit表中的数据写入到fruit_mr表中
package com.buwenbuhuo.hbase.mr;

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 卜温不火
 * @create 2020-05-12 19:32
 * com.buwenbuhuo.hbase.mr - the name of the target package where the new class or interface will be created.
 * hbase0512 - the name of the current project.
 */
public class WriteReducer 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);
        }
    }
}
  • 3. 构建Driver用于组装运行Job任务
package com.buwenbuhuo.hbase.mr;

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.HRegionPartitioner;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.mapreduce.Job;

import java.io.IOException;

/**
 * @author 卜温不火
 * @create 2020-05-12 19:32
 * com.buwenbuhuo.hbase.mr - the name of the target package where the new class or interface will be created.
 * hbase0512 - the name of the current project.
 */
public class Driver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "hadoop002,hadoop003,hadoop004");
        conf.set("hbase.zookeeper.property.clientPort", "2181");

        Job job = Job.getInstance();

        job.setJarByClass(Driver.class);

        Scan scan = new Scan();

        TableMapReduceUtil.initTableMapperJob(
                "fruit",
                scan,
                ReadMapper.class,
                ImmutableBytesWritable.class,
                Put.class,
                job
        );

        job.setNumReduceTasks(100);

        TableMapReduceUtil.initTableReducerJob("fruit_mr",WriteReducer.class,job,HRegionPartitioner.class);

        job.waitForCompletion(true);

    }

}
  • 4. 打包并上传
  • 5. 创建表
hbase(main):003:0> create 'fruit_mr','info'
  • 6. 运行验证
[bigdata@hadoop002 hbase]$ hadoop jar hbase-0512-1.0-SNAPSHOT.jar com.buwenbuhuo.hbase.mr.Driver
hbase(main):005:0> scan 'fruit_mr'

3. 自定义HBase-MapReduce2

目标:实现将HDFS中的数据写入到HBase表中。

  • 1. 构建ReadFruitFromHDFSMapper于读取HDFS中的文件数据
package com.buwenbuhuo.hbase.mr2;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @author 卜温不火
 * @create 2020-05-12 22:41
 * com.buwenbuhuo.hbase.mr2 - the name of the target package where the new class or interface will be created.
 * hbase0512 - the name of the current project.
 */
public class ReadMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        String[] split = value.toString().split("\t");

        if (split.length <= 3){
            return;
        }

        Put put = new Put(Bytes.toBytes(split[0]));

        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"),Bytes.toBytes(split[1]));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("color"),Bytes.toBytes(split[2]));

        context.write(new ImmutableBytesWritable(Bytes.toBytes(split[0])),put);

    }
}
  • 2. 构建WriteFruitMRFromTxtReducer类
package com.buwenbuhuo.hbase.mr2;

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 卜温不火
 * @create 2020-05-12 23:09
 * com.buwenbuhuo.hbase.mr2 - the name of the target package where the new class or interface will be created.
 * hbase0512 - the name of the current project.
 */
public class WriteReducer 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);
        }

    }
}
  • 3.创建Driver
package com.buwenbuhuo.hbase.mr2;

import com.buwenbuhuo.hbase.mr.WriteReducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.HRegionPartitioner;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import java.io.IOException;

/**
 * @author 卜温不火
 * @create 2020-05-12 23:09
 * com.buwenbuhuo.hbase.mr2 - the name of the target package where the new class or interface will be created.
 * hbase0512 - the name of the current project.
 */
public class Driver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "hadoop002,hadoop003,hadoop004");
        Job job = Job.getInstance(conf);

        job.setJarByClass(Driver.class);

        job.setMapperClass(ReadMapper.class);

        job.setMapOutputKeyClass(ImmutableBytesWritable.class);
        job.setMapOutputValueClass(Put.class);

        FileInputFormat.setInputPaths(job,new Path("/input_fruit"));

        job.setNumReduceTasks(10);

        TableMapReduceUtil.initTableReducerJob("fruit_mr", WriteReducer.class,job, HRegionPartitioner.class);

        job.waitForCompletion(true);
    }

}
  • 4. 打包上传
  • 5.测试运行
[bigdata@hadoop002 hbase]$ hadoop jar hbase-0512-1.0-SNAPSHOT.jar com.buwenbuhuo.hbase.mr2.Driver

  本次的分享就到这里了

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • MapReduce快速入门系列(13) | MapReduce之reduce端join与map端join算法实现

      Map端的主要工作:为来自不同表或文件的key/value对,打标签以区别不同来源的记录。然后用连接字段作为key,其余部分和新加的标志作为val,最后进行...

    不温卜火
  • MapReduce快速入门系列(2) | 统计输出给定的文本文档每一个单词出现的总次数

    下面的跟之前使用API一样,我们同样需要在IDEA中使用JAVA代码来书写MapReduce。这时候我们需要新建一个一个Maven工程

    不温卜火
  • MapReduce快速入门系列(14) | MapReduce之计数器应用及简单的数据清洗(ETL)

      Hadoop为每个作业维护若干内置计数器,以描述多项指标。   比如说,某些计数器记录已处理的字节数和记录数,使用户可监控已处理的输入数据量和已产生的输出...

    不温卜火
  • HBase Java API 03:HBase与MapReduce整合

    编写MapReduce程序,把"student"表中"info"列族下的"name"那一列抽取出来,存入新HBase表"student_extract"中,要求...

    CoderJed
  • Mapreduce和HBase新版本整合之WordCount计数案例

    先计数单词数量存到hdfs文件上,这个是以前的就做过的 package com.my.myhnase.mapreduce; import java.io.IO...

    汤高
  • HBase新的客户端接口

    最近学习接触HBase的东西,看了《Habase in Action》,但里面关于HBase接口都是过时的接口,以下为HBase新的客户端接口:

    chaplinthink
  • HBase整合MapReduce之建立HBase索引

    HBase索引主要用于提高Hbase中表数据的访问速度,有效的避免了全表扫描,HBase中的表根据行健被分成了多个Regions,通常一个region的一行都会...

    汤高
  • Hadoop Mapper 阶段将数据直接从 HDFS 导入 Hbase

    数据源格式如下: 20130512 1 -1 -1 13802 1 2013-05-12 07:26:22 20130512 1 -1 -1 13802 ...

    用户1177713
  • 大数据应用之HBase数据插入性能优化实测教程

    大家在使用HBase的过程中,总是面临性能优化的问题,本文从HBase客户端参数设置的角度,研究HBase客户端数据批量插入性能优化的问题。事实胜于雄辩,数据比...

    数据饕餮
  • HBase 2.0 协处理器实现 ES 数据同步

    在正式进行讲述实现之前,我觉得有必要说一下出发点。团队期初数据都是基于 HBase+Phoenix 这样架构进行持久化。随着业务的复杂性增加,对部分表的查询效率...

    迹_Jason

扫码关注云+社区

领取腾讯云代金券