专栏首页LhWorld哥陪你聊算法Hbase篇--Hbase和MapReduce结合Api

Hbase篇--Hbase和MapReduce结合Api

一.前述

Mapreduce可以自定义Inputforma对象和OutPutformat对象,所以原理上Mapreduce可以和任意输入源结合。

二.步骤

将结果写会到hbase中去。

 2.1 Main函数

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;


/**
 * 分析hdfs 文本  统计单词数量
 * 结果输出到 hbase表
 * create 'wc','cf'
 * rowkey: 单词        cf:count=单词数量
 * @author root
 *
 */
public class WCDemo {

    /**
     * 
     * wc
     * 数据hbase表    rowkey  cell存放文本
     * 结果输出到 hbase表
     * 
     */

    public static void main(String[] args) throws Exception {
        
        Configuration conf = new Configuration();
        
        conf.set("fs.defaultFS", "hdfs://node1:8020");//设置hdfs集群nameservices名称
        conf.set("hbase.zookeeper.quorum", "node4");
        
        Job job = Job.getInstance(conf);
        
        job.setJarByClass(WCDemo.class);
        
        job.setMapperClass(WCMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        
//        job.setReducerClass();
        
        //addDependencyJars  本地方式运行: 设置为false
//        TableMapReduceUtil.initTableReducerJob("wc", WCReducer.class, job);
        TableMapReduceUtil.initTableReducerJob("wc",WCReducer.class, job,
                null, null, null, null, false);
        
        Path path = new Path("/user/wc");
        FileInputFormat.addInputPath(job, path);
        
        boolean flag = job.waitForCompletion(true);
        if(flag) {
            System.out.println("success~~");
        }
    }
    
}

2.2 Mapper函数(和正常的Mapper没啥区别)

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class WCMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String[] words = value.toString().split(" ");
        
        for (String w : words) {
            context.write(new Text(w), new IntWritable(1));
        }
    }
}

2.3 Reduce函数(主要是把Put对象写出去)

import java.io.IOException;

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.IntWritable;
import org.apache.hadoop.io.Text;

public class WCReducer extends
        TableReducer<Text, IntWritable, ImmutableBytesWritable> {

    @Override
    protected void reduce(Text text, Iterable<IntWritable> iterable,
            Context context) throws IOException, InterruptedException {
        
        int sum = 0;
        
        for (IntWritable i : iterable) {
            sum += i.get();
        }
        
        Put put = new Put(text.toString().getBytes());
        put.add("cf".getBytes(), "count".getBytes(), (sum+"").getBytes());
        
        context.write(null, put);
    }
}

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

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 【Spark篇】---SparkStream初始与应用

    SparkStreaming是流式处理框架,是Spark API的扩展,支持可扩展、高吞吐量、容错的实时数据流处理,实时数据的来源可以是:Kafka, Flum...

    LhWorld哥陪你聊算法
  • 【Spark篇】---SparkStreaming算子操作transform和updateStateByKey

    今天分享一篇SparkStreaming常用的算子transform和updateStateByKey。

    LhWorld哥陪你聊算法
  • 【Spark篇】---Spark中Action算子

    Action类算子也是一类算子(函数)叫做行动算子,如foreach,collect,count等。Transformations类算子是延迟执行,Action...

    LhWorld哥陪你聊算法
  • 使用Java Api 操作HDFS

    如题 我就是一个标题党 就是使用JavaApi操作HDFS,使用的是MAVEN,操作的环境是Linux

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

    不温卜火
  • 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
  • Mapreduce和HBase新版本整合之WordCount计数案例

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

    汤高
  • HBase新的客户端接口

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

    chaplinthink
  • HBase Java API 03:HBase与MapReduce整合

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

    CoderJed

扫码关注云+社区

领取腾讯云代金券