【Hadoop】Hadoop MR 自定义分组 Partition机制

1、概念

 

2、Hadoop默认分组机制--所有的Key分到一个组，一个Reduce任务处理

3、代码示例

FlowBean

package com.ares.hadoop.mr.flowgroup;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;import org.apache.hadoop.io.WritableComparable;public class FlowBean implements WritableComparable<FlowBean>{private String phoneNum;private long upFlow;private long downFlow;private long sumFlow;public FlowBean() {// TODO Auto-generated constructor stub
    }    
//    public FlowBean(String phoneNum, long upFlow, long downFlow, long sumFlow) {
//        super();
//        this.phoneNum = phoneNum;
//        this.upFlow = upFlow;
//        this.downFlow = downFlow;
//        this.sumFlow = sumFlow;
//    }public String getPhoneNum() {return phoneNum;}public void setPhoneNum(String phoneNum) {this.phoneNum = phoneNum;}public long getUpFlow() {return upFlow;}public void setUpFlow(long upFlow) {this.upFlow = upFlow;}public long getDownFlow() {return downFlow;}public void setDownFlow(long downFlow) {this.downFlow = downFlow;}public long getSumFlow() {return sumFlow;}public void setSumFlow(long sumFlow) {this.sumFlow = sumFlow;}@Overridepublic void readFields(DataInput in) throws IOException {// TODO Auto-generated method stubphoneNum = in.readUTF();upFlow = in.readLong();downFlow = in.readLong();sumFlow = in.readLong();}@Overridepublic void write(DataOutput out) throws IOException {// TODO Auto-generated method stubout.writeUTF(phoneNum);out.writeLong(upFlow);out.writeLong(downFlow);out.writeLong(sumFlow);}@Overridepublic String toString() {return "" + phoneNum + "\t" + upFlow + "\t" + downFlow + "\t" + sumFlow;}@Overridepublic int compareTo(FlowBean o) {// TODO Auto-generated method stubreturn sumFlow>o.getSumFlow()?-1:1;}}

FlowGroup

package com.ares.hadoop.mr.flowgroup;import java.io.IOException;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;import com.ares.hadoop.mr.exception.LineException;
import com.ares.hadoop.mr.flowgroup.FlowBean;;public class FlowGroup extends Configured implements Tool {private static final Logger LOGGER = Logger.getLogger(FlowGroup.class);enum Counter {LINESKIP}public static class FlowGroupMapper extends Mapper<LongWritable, Text, Text, FlowBean> {private String line;private int length;private final static char separator = '\t';private String phoneNum;private long upFlow;private long downFlow;//private long sumFlow;private Text text = new Text();private FlowBean flowBean = new FlowBean();@Overrideprotected void map(LongWritable key,Text value,Mapper<LongWritable, Text, Text, FlowBean>.Context context)throws IOException, InterruptedException {// TODO Auto-generated method stub//super.map(key, value, context);
            String errMsg;try {line = value.toString();String[] fields = StringUtils.split(line, separator);length = fields.length;if (length != 11) {throw new LineException(key.get() + ", " + line + " LENGTH INVALID, IGNORE...");}phoneNum = fields[1];upFlow = Long.parseLong(fields[length-3]);downFlow = Long.parseLong(fields[length-2]);//sumFlow = upFlow + downFlow;
                text.set(phoneNum);flowBean.setPhoneNum(phoneNum);flowBean.setUpFlow(upFlow);flowBean.setDownFlow(downFlow);//flowBean.setSumFlow(sumFlow);
                context.write(text, flowBean);} catch (LineException e) {// TODO: handle exception
                LOGGER.error(e);System.out.println(e);context.getCounter(Counter.LINESKIP).increment(1);return;} catch (NumberFormatException e) {// TODO: handle exceptionerrMsg = key.get() + ", " + line + " FLOW DATA INVALID, IGNORE...";LOGGER.error(errMsg);System.out.println(errMsg);context.getCounter(Counter.LINESKIP).increment(1);return;} catch (Exception e) {// TODO: handle exception
                LOGGER.error(e);System.out.println(e);context.getCounter(Counter.LINESKIP).increment(1);return;}            }}public static class FlowGroupReducer extends Reducer<Text, FlowBean, FlowBean, NullWritable> {private FlowBean flowBean = new FlowBean();@Overrideprotected void reduce(Text key,Iterable<FlowBean> values,Reducer<Text, FlowBean, FlowBean, NullWritable>.Context context)throws IOException, InterruptedException {// TODO Auto-generated method stub//super.reduce(arg0, arg1, arg2);long upFlowCounter = 0;long downFlowCounter = 0;for (FlowBean flowBean : values) {upFlowCounter += flowBean.getUpFlow();downFlowCounter += flowBean.getDownFlow();}flowBean.setPhoneNum(key.toString());flowBean.setUpFlow(upFlowCounter);flowBean.setDownFlow(downFlowCounter);flowBean.setSumFlow(upFlowCounter + downFlowCounter);context.write(flowBean, NullWritable.get());}}@Overridepublic int run(String[] args) throws Exception {// TODO Auto-generated method stubString errMsg = "FlowGroup: TEST STARTED...";LOGGER.debug(errMsg);System.out.println(errMsg);Configuration conf = new Configuration();//FOR Eclipse JVM Debug  //conf.set("mapreduce.job.jar", "flowsum.jar");Job job = Job.getInstance(conf);// JOB NAMEjob.setJobName("FlowGroup");// JOB MAPPER & REDUCERjob.setJarByClass(FlowGroup.class);job.setMapperClass(FlowGroupMapper.class);job.setReducerClass(FlowGroupReducer.class);// JOB PARTITIONjob.setPartitionerClass(FlowGroupPartition.class);// JOB REDUCE TASK NUMBERjob.setNumReduceTasks(5);// MAP & REDUCEjob.setOutputKeyClass(FlowBean.class);job.setOutputValueClass(NullWritable.class);// MAPjob.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(FlowBean.class);// JOB INPUT & OUTPUT PATH//FileInputFormat.addInputPath(job, new Path(args[0]));FileInputFormat.setInputPaths(job, args[1]);FileOutputFormat.setOutputPath(job, new Path(args[2]));// VERBOSE OUTPUTif (job.waitForCompletion(true)) {errMsg = "FlowGroup: TEST SUCCESSFULLY...";LOGGER.debug(errMsg);System.out.println(errMsg);return 0;} else {errMsg = "FlowGroup: TEST FAILED...";LOGGER.debug(errMsg);System.out.println(errMsg);return 1;}            }public static void main(String[] args) throws Exception {if (args.length != 3) {String errMsg = "FlowGroup: ARGUMENTS ERROR";LOGGER.error(errMsg);System.out.println(errMsg);System.exit(-1);}int result = ToolRunner.run(new Configuration(), new FlowGroup(), args);System.exit(result);}
}

FlowGroupPartition

package com.ares.hadoop.mr.flowgroup;import java.util.HashMap;import org.apache.hadoop.mapreduce.Partitioner;public class FlowGroupPartition<KEY, VALUE> extends Partitioner<KEY, VALUE>{private static HashMap<String, Integer> groupMap = new HashMap<String, Integer>();static {groupMap.put("135", 0);groupMap.put("136", 1);groupMap.put("137", 2);groupMap.put("138", 3);}@Overridepublic int getPartition(KEY key, VALUE value, int numPartitions) {// TODO Auto-generated method stubreturn (groupMap.get(key.toString().substring(0, 3)) == null)?4:groupMap.get(key.toString().substring(0, 3));}}

 

转载于:https://www.cnblogs.com/junneyang/p/5849257.html

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/372354.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Spring Framework 3.2 M1发布

Spring Framework 3.2 M1发布

SpringSource刚刚宣布了针对Spring 3.2的第一个里程碑版本。 现在可以从SpringSource存储库&#xff08;位于http://repo.springsource.org/&#xff09;获得新版本。 查看有关通过Maven 解决这些工件的快速教程 。 此版本包括&#xff1a; 最初支持异步Controller方法 早期…
阅读更多...
两种动态SQL

两种动态SQL

参考&#xff1a;http://www.cnblogs.com/wanyuan8/archive/2011/11/09/2243483.htmlhttp://www.cnblogs.com/xbf321/archive/2008/11/02/1325067.html 两种动态SQL　　1. EXEC (sql) 　　2. EXEC sp_executesql 性能&#xff1a;sp_executesql提供了输入输出接口&#xff0c;更…
阅读更多...
mysql查询含有某个值的表_MYSQL查询数据表中某个字段包含某个数值

mysql查询含有某个值的表_MYSQL查询数据表中某个字段包含某个数值

当某个字段中字符串是"1,2,3,4,5,6"或者"123456" 查询数据表中某个字段是否包含某个值 1:模糊查询 使用like select * from 表 where 字段 like %1%; 2:函数查找 find_in_set(str，数组) select * from 表 where find_in_set(1,字段); 注意：mysql字符串…
阅读更多...
android学习笔记35——AnimationDrawable资源

android学习笔记35——AnimationDrawable资源

AnimationDrawable资源 AnimationDrawable&#xff0c;代表一个动画。 android既支持传统的逐帧动画(类似于电影方式&#xff0c;一张图片一张图片的切换)&#xff0c;也支持通过平移、变换计算出来的补间动画、属性动画。 下面以补间动画为例&#xff0c;介绍如何定义Animatio…
阅读更多...
RESTEasy教程第2部分：Spring集成

RESTEasy教程第2部分：Spring集成

RESTEasy提供了对Spring集成的支持&#xff0c;这使我们能够将Spring bean作为RESTful WebServices公开。 步骤&#xff03;1&#xff1a;使用Maven配置RESTEasy Spring依赖项。 <project xmlnshttp:maven.apache.orgPOM4.0.0 xmlns:xsihttp:www.w3.org2001XMLSchema-insta…
阅读更多...
java RSA 加签验签【转】

java RSA 加签验签【转】

引用自: http://blog.csdn.net/wangqiuyun/article/details/42143957/ java RSA 加签验签 package com.testdemo.core.service.impl.alipay;import java.security.KeyFactory; import java.security.PrivateKey; import java.security.PublicKey; import java.security.spec.PK…
阅读更多...
mysql启动时执行sql server_常见 mysql 启动、运行.sql 文件错误处理

mysql启动时执行sql server_常见 mysql 启动、运行.sql 文件错误处理

1、mysql 启动错误处理查看 log&#xff1a;Mac: /usr/local/var/mysql/lizhendeMacBook-Pro.local.err根据 log 针对性的进行调整&#xff0c;包治百病2、Mysql Incorrect datetime value问题描述&#xff1a;低版本的 mysql 中&#xff0c;数据库转储 sql 文件。导入到高版本…
阅读更多...
带有谓词的Java中的函数样式-第2部分

带有谓词的Java中的函数样式-第2部分

在本文的第一部分中&#xff0c;我们介绍了谓词&#xff0c;这些谓词通过具有返回true或false的单个方法的简单接口&#xff0c;为Java等面向对象的语言带来了函数式编程的某些好处。 在第二部分和最后一部分中&#xff0c;我们将介绍一些更高级的概念&#xff0c;以使您的谓词…
阅读更多...
Devxtreme 显示Master-Detail数据列表, 数据显示颜色

Devxtreme 显示Master-Detail数据列表, 数据显示颜色

1 ////刷新3/4簇Grid2 //function GetClusterGrid(id, coverageId, clusterId) {3 4 // var region getRegionCityName();5 // $.ajax({6 // type: "POST",7 // url: "fast_index_overview.aspx/GetClusterGrid&q…
阅读更多...
mysql 排序去重复_php mysql 过滤重复记录并排序

mysql 排序去重复_php mysql 过滤重复记录并排序

table1idname1a2b3ctable2idnamecont1aaa2bbb3aaaaaSELECT*,count(distincttable2.name)FROMtable1,table2WHEREtable1.nametable2.nameGROUPBYtable2.nameORDERBYtable2.idDESC";重复...table1id name1 a2 b3 ctable2id name cont1 a aa2 b bb3 a aaaaSELECT *,count(dis…
阅读更多...
Java EE 6测试第I部分– EJB 3.1可嵌入API

Java EE 6测试第I部分– EJB 3.1可嵌入API

我们从Enterprise JavaBeans开发人员那里听到的最常见的请求之一就是需要改进的单元/集成测试支持。 EJB 3.1 Specification引入了EJB 3.1 Embeddable API&#xff0c;用于在Java SE环境中执行EJB组件。 与传统的基于Java EE服务器的执行不同&#xff0c;可嵌入式用法允许客户端…
阅读更多...
Flume 中文入门手冊

Flume 中文入门手冊

原文&#xff1a;https://cwiki.apache.org/confluence/display/FLUME/GettingStarted 什么是 Flume NG? Flume NG 旨在比起 Flume OG 变得明显更简单。更小。更easy部署。在这样的情况下&#xff0c;我们不提交Flume NG 到 Flume OG 的后向兼容。当前。我们期待来自感兴趣測试…
阅读更多...
原生JavaScript+CSS3实现移动端滑块效果

原生JavaScript+CSS3实现移动端滑块效果

在做web页面时&#xff0c;无论PC端还是移动端&#xff0c;我们会遇到滑块这样的效果&#xff0c;可能我们往往会想着去网上找插件&#xff0c;其实这个效果非常的简单&#xff0c;插件代码的的代码往往过于臃肿&#xff0c;不如自己动手&#xff0c;自给自足。首先看一下效果图…
阅读更多...
mysql的连接名是哪个文件_mysql连接名是什么

mysql的连接名是哪个文件_mysql连接名是什么

{"moduleinfo":{"card_count":[{"count_phone":1,"count":1}],"search_count":[{"count_phone":4,"count":4}]},"card":[{"des":"阿里云数据库专家保驾护航&#xff0c;为用户…
阅读更多...
Activiti绩效对决

Activiti绩效对决

每个人在学习Activiti时都会一直问到的问题&#xff0c;与软件开发本身一样古老&#xff1a;“它如何执行&#xff1f;”。 到现在为止&#xff0c;当您问我同样的问题时&#xff0c;我将告诉您Activiti如何以各种可能的方式最小化数据库访问&#xff0c;如何将流程结构分解为“…
阅读更多...
怎么使用CKEDITOR

怎么使用CKEDITOR

出于工作需求&#xff0c;自己在网上找了个文本编辑器控件, 网址是http://ckeditor.com/ 怎么使用&#xff1f; 先插入脚本<script type"text/javascript" src"*/ckeditor.js"></script>, 然后&#xff0c;在自己的脚本里调用CKEDITOR.replac…
阅读更多...
centos 打开pdo_mysql_centos中添加php扩展pdo_mysql步骤

centos 打开pdo_mysql_centos中添加php扩展pdo_mysql步骤

pdo_mysql是php中一个mysql连接类了&#xff0c;我们可以直接使用pdo_mysql来操作数据库这样自己可以不需要写数据库操作类了&#xff0c;下面来介绍在centos中安装pdo_mysql扩展的步骤。本文内容是以 CentOS 为例&#xff0c;红帽系列的 Linux 方法应该都是如此&#xff0c;下…
阅读更多...
Java线程死锁–案例研究

Java线程死锁–案例研究

本文将描述从在IBM JVM 1.6上运行的Weblogic 11g生产系统中观察到的最新Java死锁问题的完整根本原因分析。 此案例研究还将证明掌握线程转储分析技能的重要性&#xff1b; 包括用于IBM JVM Thread Dump格式。 环境规格 – Java EE服务器&#xff1a;Oracle Weblogic Server 1…
阅读更多...
bzoj1968: [Ahoi2005]COMMON 约数研究

bzoj1968: [Ahoi2005]COMMON 约数研究

水题。。。 #include<cstdio> #include<cstring> #include<iostream> #include<algorithm> using namespace std; #define rep(i,s,t) for(int is;i<t;i) int main(){int ans0,n;scanf("%d",&n);rep(i,1,n) ansn/i;printf("%d\n…
阅读更多...
题目1457：非常可乐(广度优先遍历BFS)

题目1457：非常可乐(广度优先遍历BFS)

题目链接&#xff1a;http://ac.jobdu.com/problem.php?pid1457 详解链接&#xff1a;https://github.com/zpfbuaa/JobduInCPlusPlus 参考代码&#xff1a; // // 1457 非常可乐.cpp // Jobdu // // Created by PengFei_Zheng on 22/04/2017. // Copyright © 2017 Pe…
阅读更多...
最新文章

Copyright @ 2022~2023