HBase与Mapreduce

HBase与Mapreduce

一、统计行数案例

$HBASE_HOME/bin/hbase org.apache.hadoop.hbase.mapreduce.RowCounter ‘tablename’

二、将HDFS中数据导入到hbase

1、准备数据

student.txt

0001,小1,21
0002,小2,22
0003,小3,23
0004,小4,24
0005,小5,25
0006,小6,21
0007,小7,22

2、创建hbase表

hbase(main):003:0> create 'student','info'

3、在 HDFS 中创建 /data文件夹并上传 student.txt 文件

 hdfs dfs -put student.txt /data

4、执行 MapReduce 到 HBase 的 fruit 表中

yarn jar /software/hbase-1.3.1/lib/hbase-server-1.3.1.jar importtsv -Dimporttsv.separator="," \
-Dimporttsv.columns=HBASE_ROW_KEY,info:name,info:age student \
hdfs://node01:9000/data/student.txt

5、结果

三、使用代码将HDFS数据导入HBase

1、maven

    <dependency>
        <groupId>org.apache.hbase</groupId>
        <artifactId>hbase-server</artifactId>
        <version>1.3.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hbase</groupId>
        <artifactId>hbase-client</artifactId>
        <version>1.3.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>2.7.4</version>
    </dependency>

2、编写mapper

class ReadHDFSData2HbaseMapper extends Mapper<LongWritable,
            Text, ImmutableBytesWritable, Put> {
        @Override
        protected void map(LongWritable key, Text value, Context
                context) throws IOException, InterruptedException {
          //从 HDFS 中读取的数据
            String lineValue = value.toString();
          //读取出来的每行数据使用\t 进行分割，存于 String 数组
            String[] values = lineValue.split(",");
            System.out.println(lineValue);
          //根据数据中值的含义取值
            String rowKey = values[0];
            String name = values[1];
            String age = values[2];
          //初始化 rowKey
            ImmutableBytesWritable rowKeyWritable = new
                    ImmutableBytesWritable(Bytes.toBytes(rowKey));
            //初始化 put 对象
            Put put = new Put(Bytes.toBytes(rowKey));
          //参数分别:列族、列、值
            put.add(Bytes.toBytes("info"), Bytes.toBytes("name"),
                    Bytes.toBytes(name));
            put.add(Bytes.toBytes("info"), Bytes.toBytes("age"),
                    Bytes.toBytes(age));
            context.write(rowKeyWritable, put);
        }
    }

3、编写reduce

class ReadHDFSData2HbaseReducer extends
            TableReducer<ImmutableBytesWritable, Put, NullWritable> {
        @Override
        protected void reduce(ImmutableBytesWritable key, Iterable<Put>
                values, Context context) throws IOException, InterruptedException {
            //读出来的每一行数据写入到 hbase 表中
            for (Put put : values) {
                context.write(NullWritable.get(), put);
            }
        }
    }

4、编写driver

public class ReadHDFSDataToHbaseMR  extends Configured implements Tool {
    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        int status = ToolRunner.run(conf, new ReadHDFSDataToHbaseMR(),
                args);
        System.exit(status);
    }

    public int run(String[] args) throws Exception {
        //得到 Configuration
        Configuration conf = this.getConf();
        //创建 Job 任务
        Job job = Job.getInstance(conf, this.getClass().getSimpleName());
        job.setJarByClass(ReadHDFSDataToHbaseMR.class);
        Path  inPath  =  new Path("hdfs://node02:9000/data/student.txt");
        FileInputFormat.addInputPath(job, inPath);
        //设置 Mapper
        job.setMapperClass(ReadHDFSData2HbaseMapper.class);
        job.setMapOutputKeyClass(ImmutableBytesWritable.class);
        job.setMapOutputValueClass(Put.class);
        //设置 Reducer
        TableMapReduceUtil.initTableReducerJob("student",
                ReadHDFSData2HbaseReducer.class, job);
        //设置 Reduce 数量，最少 1 个
        job.setNumReduceTasks(1);
        boolean isSuccess = job.waitForCompletion(true);
        if(!isSuccess){
            throw new IOException("Job running with error");
        }
        return isSuccess ? 0 : 1;
    }
}

5、打包运行

hadoop jar HbaseDemo-1.0-SNAPSHOT.jar com.mc.hbase.mr.ReadHDFSDataToHbaseMR