hadoop2.2.0 MapReduce分区

 package com.my.hadoop.mapreduce.partition;

import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class ParCount {

public static class ParMap extends Mapper<LongWritable, Text, Text, InfoBean>{
        private Text key = new Text();
        @Override
        public void map(LongWritable key, Text value, Context context) throws java.io.IOException ,InterruptedException {
            String[] fields = value.toString().split("\t");
            String telNo = fields[1];
            long upPayLoad = Long.parseLong(fields[8]);
            long downPayLoad = Long.parseLong(fields[9]);
            InfoBean bean = new InfoBean(telNo, upPayLoad, downPayLoad);
            this.key.set(telNo);
            context.write(this.key, bean);
        }
    }
    
    public static class ParReduce extends Reducer<Text, InfoBean, Text, InfoBean>{
        @Override
        public void reduce(Text key, java.lang.Iterable<InfoBean> value, org.apache.hadoop.mapreduce.Reducer<Text,InfoBean,Text,InfoBean>.Context context) throws java.io.IOException ,InterruptedException {
            long up_sum = 0;
            long down_sum = 0;
            for (InfoBean bean : value) {
                up_sum += bean.getUpPayLoad();
                down_sum += bean.getDownPayLoad();
            }
            InfoBean bean = new InfoBean("", up_sum, down_sum);
            context.write(key, bean);
        }
    }
    
    /**
     * 分区,参数为Map的输出
     * @author yao
     *
     */
    public static class MyPar extends Partitioner<Text, InfoBean>{

private static Map<String, Integer> parFlag = new HashMap<String, Integer>();
        static {
            parFlag.put("135", 1);
            parFlag.put("136", 1);
            parFlag.put("137", 1);
            parFlag.put("138", 1);
            parFlag.put("139", 1);
            parFlag.put("150", 2);
            parFlag.put("159", 2);
            parFlag.put("182", 3);
            parFlag.put("183", 3);
        }
        
        @Override
        public int getPartition(Text key, InfoBean value, int arg2) {
            String telNo = key.toString().substring(0, 3);
            Integer code = parFlag.get(telNo);
            if (code == null) {
                code = 0;
            }
            
            return code;
        }
        
    }
    
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, ParCount.class.getSimpleName());;
        job.setJarByClass(ParCount.class);
        
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        job.setMapperClass(ParMap.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(InfoBean.class);
        
        job.setPartitionerClass(MyPar.class);                                        //指定自定义的分区类
        job.setNumReduceTasks(4);                   //需要根据分区的数量设置Reducer数量,多了会出现空文件,少了会报错
        
        job.setReducerClass(ParReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(InfoBean.class);
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        System.exit(job.waitForCompletion(true) ? 0 : 1);
        
    }

}

时间: 2024-10-07 16:21:18

hadoop2.2.0 MapReduce分区的相关文章

hadoop2.2.0 MapReduce的序列化

package com.my.hadoop.mapreduce.dataformat; import java.io.DataInput;import java.io.DataOutput;import java.io.IOException; import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.LongWritable;import or

hadoop2.2.0 MapReduce求和并排序

javabean必须实现WritableComparable接口,并实现该接口的序列化,反序列话和比较方法 package com.my.hadoop.mapreduce.sort; import java.io.DataInput;import java.io.DataOutput;import java.io.IOException; import org.apache.hadoop.io.WritableComparable; public class InfoBean implement

Hadoop2.6.0学习笔记(七)MapReduce分区

鲁春利的工作笔记,谁说程序员不能有文艺范? MapReduce中map task任务的数量是由spli分片决定,那么reduce task的数量由什么来确定的呢?就是这里要讨论的MapReduce分区.默认情况下,MapReduce中使用的是HashPartitioner. /** Partition keys by their {@link Object#hashCode()}. */ public class HashPartitioner<K, V> extends Partitione

Hadoop2.2.0伪分布式之MapReduce简介

一概念. mapReduce是分布式计算模型.注:在hadoop2.x中MapReduce运行在yarn上,yarn支持多种运算模型.storm.spark等等,任何运行在JVM上的程序都可以运行在yarn上. MR有两个阶段组成,Map和Reduce,用户只需要实现Map()和reduce()两个函数(且这两个函数的输入和输出均是key -value的形式)即可实现分布式计算.代码示例略. MapReduce设计框架: 在1.0中:,管理者:Job Tracker:被管理者:Task Trac

Hadoop2.6.0运行mapreduce之Uber模式验证

前言 在有些情况下,运行于Hadoop集群上的一些mapreduce作业本身的数据量并不是很大,如果此时的任务分片很多,那么为每个map任务或者reduce任务频繁创建Container,势必会增加Hadoop集群的资源消耗,并且因为创建分配Container本身的开销,还会增加这些任务的运行时延.如果能将这些小任务都放入少量的Container中执行,将会解决这些问题.好在Hadoop本身已经提供了这种功能,只需要我们理解其原理,并应用它. Uber运行模式就是解决此类问题的现成解决方案.本文

Hadoop-2.2.0中文文档——MapReduce 下一代 -——集群配置

目的 这份文档描写叙述了怎样安装.配置和管理从几个节点到有数千个节点的Hadoop集群. 玩的话,你可能想先在单机上安装.(看单节点配置). 准备 从Apache镜像上下载一个Hadoop的稳定版本号. 安装 安装一个Hadoop集群,一般包含分发软件到全部集群中的机器上或者是安装RPMs. 一般地,集群中的一台机器被唯一地设计成NameNode,还有一台机器被设置成ResourceManager.这是master(主). 集群中剩下的机器作为DataNode 和 NodeManager.这些是

Hadoop-2.2.0中文文档—— MapReduce 下一代 - Encrypted Shuffle

简介 Encrypted Shuffle capability (加密洗牌功能?)允许用HTTPS 和 可选的客户端验证 (也称作双向的 HTTPS, 或有客户端证书的 HTTPS) 去加密 MapReduce shuffle.它包括: 在HTTP 和 HTTPS 之间绑定 shuffle 的一个 Hadoop 配置 用来指定 keystore 和 truststore 属性的Hadoop配置(位置,类型,密码) 用于 shuffle 服务和reducer任务去取 shuffle 数据. 在集群

Hadoop-2.2.0中文文档—— MapReduce 下一代 -- 公平调度器

目的 此文档描述了 FairScheduler, Hadoop 的一个可插入式的调度器,允许 YARN 应用在一个大集群中公平地共享资源. 简介 公平调度是一种分配资源给应用的方法,以致到最后,平均上所有应用获得相等的资源.  Hadoop NextGen 能够调度多种类型的资源.默认的, Fair Scheduler 仅以内存为基础作公平调度决策.可以用Ghodsi 等开发的 Dominant Resource Fairness 概念配置调度内存和CPU.仅有一个应用运行时,这个应用使用整个集

记Hadoop2.5.0线上mapreduce任务执行map任务划分的一次问题解决

前言 近日在线上发现有些mapreduce作业的执行时间很长,我们需要解决这个问题.输入文件的大小是5G,采用了lzo压缩,整个集群的默认block大小是128M.本文将详细描述这次线上问题的排查过程. 现象 线上有一个脚本,为了便于展示,我将这个脚本重新copy了一份并重命名为zzz.这个脚本实际是使用Hadoop streaming运行一个mapreduce任务,在线上执行它的部分输出内容如下: 可以看到map任务划分为1个.这个执行过程十分漫长,我将中间的一些信息省略,map与reduce