首页 文章 精选 留言 我的

使用Mapreduce案例编写用于统计文本中单词出现的次数的案例、mapreduce本地运行等，Combiner使用及其相关的知识，流量统计...

2017-06-01 684

工程结构：

在整个案例过程中，代码如下：

WordCountMapper的代码如下：

package cn.toto.bigdata.mr.wc;

 

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;

 

/**

 * 这里的Mapper是hadoop-mapreduce-client-core-2.8.0.jar中的内容

 * Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>

 * KEYIN     :是指框架读取到的数据的key的类型，在默认的InputFormat下，读到的key是一行文本的起始偏移量，所以key的类型是Long

 * VALUEIN   :是指框架读取到的数据的value的类型，在默认的InputFormat下，读到的value是一行文本的内容，所以value的类型是String

 * KEYOUT    :是指用户自定义逻辑方法返回的数据中key的类型，由用户业务逻辑决定，在此wordcount程序中，我们输出的key是单词，所以是String

 * VALUEOUT  :是指用户自定义逻辑方法返回的数据中value的类型，由用户业务逻辑决定,在此wordcount程序中，我们输出的value是单词的数量，所以是Integer

 *

 * 但是，String,Long等jdk中自带的数据类型，在序列化是，效率比较低，hadoop为了提高序列化效率，自定义了一套序列化框架，

 * 所以，在hadoop的程序中，如果该数据需要进行序列化(写磁盘，或者网络传输)，就一定要用实现了hadoop序列化框架的数据类型

 *

 * Long       ----> LongWritable

 * String     ----> Text

 * Integer    ----> IntWritable

 * Null       ----> NullWritable

 */

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

 

         /**

          * 这就是mapreduce框架中一个主体运行程序MapTask所要调用的用户业务逻辑方法

          * MapTask会驱动InputFormat去读取数据（keyIN,VALUEIN），每读到一个KV对，就传入这个用户写的map方法中调用一次

          * 在默认的inputformat实现中，此处的一个key就是一行的起始偏移量，value就是一行的内容

          */

         @Override

         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)

                            throws IOException, InterruptedException {

                  

                   String line = value.toString();

                   String[] words = line.split(" ");

                   for (String word : words) {

                            context.write(new Text(word), new IntWritable(1));

                   }

         }

}

 

WordCountReducer的代码如下：

package cn.toto.bigdata.mr.wc;

 

import java.io.IOException;

 

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

 

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

 

         /** reducetask在调我们写的reduce方法

                   reducetask应该收到了前一阶段（map阶段）中所有maptask输出的数据中的一部分

                   （数据的key.hashcode%reducetask数==本reductask号）

                   reducetask将这些收到kv数据拿来处理时，是这样调用我们的reduce方法的：

                          先将自己收到的所有的kv对按照k分组（根据k是否相同）

                          将某一组kv中的第一个kv中的k传给reduce方法的key变量，把这一组kv中所有的v用一个迭代器传给reduce方法的变量values

                    */

         @Override

         protected void reduce(Text key, Iterable<IntWritable> values,

                            Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {

                   int count = 0;

                   for(IntWritable v : values) {

                            count += v.get();

                   }

                  

                   context.write(key, new IntWritable(count));

         }

}

 

WordCountDriver的代码如下：

package cn.toto.bigdata.mr.wc;

 

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

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;

import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

 

/**

 * 本类是客户端用来指定wordcount job程序运行时所需要的很多参数：

 * 比如，指定用哪个组件作为数据读取器、数据结果输出器

 *     指定用哪个类作为map阶段的业务逻辑类，哪个类作为reduce阶段的业务逻辑类

 *     指定wordcount job程序的jar包所在路径

 *     ....

 *     运行前准备工作

 *     1、将当前的工程导出成wordcount.jar

 *     2、准备/home/toto/software/wordcount/input/a.txt 其中a.txt中的内容类似：

 *                  The true

                nobility is

                in being

                superior to

                your previous

                self guess

      3、将 /home/toto/software/wordcount通过 hadoop fs -put wordcount /wordcount 上传到hdfs中

 *    

 *     以及其他各种需要的参数

 *     hadoop jar wordcount.jar cn.toto.bigdata.mr.wc.WordCountDriver

 *     上面的命令等同：

 *     java -cp wordcount.jar cn.toto.bigdata.mr.wc.WordCountDriver

 *    

 *     上面的含义是通过hadoop jar将hadoop classpath的jar都拷贝到应用中，并且指定执行cn.toto.bigdata.mr.wc.WordCountDriver

 *     

 *     最后查看结果的方式是：hadoop fs -cat /wordcount/output/part-r-00000,通过这个命令可以查看查看到

 */

public class WordCountDriver {

 

   public static void main(String[] args) throws Exception {

      Configuration conf = new Configuration();

     

      conf.set("fs.defaultFS", "hdfs://hadoop:9000");

     

      /*conf.set("mapreduce.framework.name", "yarn");

      conf.set("yarn.resourcemanager.hostname", "mini1");*/

     

      Job job = Job.getInstance(conf);

      //告诉框架，我们的程序所在jar包的路径

      // job.setJar("c:/wordcount.jar");

      job.setJarByClass(WordCountDriver.class);

     

      //告诉框架，我们的程序所用的mapper类和reducer类

      job.setMapperClass(WordCountMapper.class);

      job.setReducerClass(WordCountReducer.class);

     

      //告诉框架，我们的mapperreducer输出的数据类型

      job.setMapOutputKeyClass(Text.class);

      job.setMapOutputValueClass(IntWritable.class);

     

      job.setOutputKeyClass(Text.class);

      job.setOutputValueClass(IntWritable.class);

     

      // 告诉框架，我们的数据读取、结果输出所用的format组件

      // TextInputFormat是mapreduce框架中内置的一种读取文本文件的输入组件

      job.setInputFormatClass(TextInputFormat.class);

      job.setOutputFormatClass(TextOutputFormat.class);

     

      // 告诉框架，我们要处理的文件在哪个路径下,注意若hdfs中已经有了/wordcount/input/这个文件，说明

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

     

      // 告诉框架，我们的处理结果要输出到哪里去

      FileOutputFormat.setOutputPath(job, new Path("/wordcount/output/"));深

     

      boolean res = job.waitForCompletion(true);

     

      System.exit(res?0:1);

   }

}

 

运行前的准备工作：

运行前准备工作

1、将当前的工程导出成wordcount.jar

2、准备/home/toto/software/wordcount/input/a.txt 其中a.txt中的内容类似：

                   The true

                nobility is

                in being

                superior to

                your previous

                self guess

   3、将 /home/toto/software/wordcount通过 hadoop fs -put wordcount /wordcount 上传到hdfs中

最后，可以执行的命令是：

hadoop jar wordcount.jar cn.toto.bigdata.mr.wc.WordCountDriver

执行后的效果如下：

 

B:使用WordCount本地运行，并且使用Combiner的案例（主要改变是在WordCountDriver中），代码如下：

package cn.toto.bigdata.mr.wc;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

/**
 * 本类是客户端用来指定wordcount job程序运行时所需要的很多参数：
 * 比如，指定用哪个组件作为数据读取器、数据结果输出器
 *     指定用哪个类作为map阶段的业务逻辑类，哪个类作为reduce阶段的业务逻辑类
 *     指定wordcount job程序的jar包所在路径
 *     ....
 *     运行前准备工作
 *     1、将当前的工程导出成wordcount.jar
 *     2、准备/home/toto/software/wordcount/input/a.txt 其中a.txt中的内容类似：
                    The true
					nobility is 
					in being 
					superior to 
					your previous 
					self guess
	   3、将 /home/toto/software/wordcount通过 hadoop fs -put wordcount /wordcount 上传到hdfs中
 *     
 *     以及其他各种需要的参数
 *     hadoop jar wordcount.jar cn.toto.bigdata.mr.wc.WordCountDriver
 *     上面的命令等同：
 *     java -cp wordcount.jar cn.toto.bigdata.mr.wc.WordCountDriver
 *     
 *     上面的含义是通过hadoop jar将hadoop classpath的jar都拷贝到应用中，并且指定执行cn.toto.bigdata.mr.wc.WordCountDriver
 *     
 *     最后查看结果的方式是：hadoop fs -cat /wordcount/output/part-r-00000,通过这个命令可以查看查看到 
 */
public class WordCountDriver {

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		
		//conf.set("fs.defaultFS", "hdfs://hadoop:9000");
		
		/*conf.set("mapreduce.framework.name", "yarn");
		conf.set("yarn.resourcemanager.hostname", "mini1");*/
		
		Job job = Job.getInstance(conf);
		//告诉框架，我们的程序所在jar包的路径
		// job.setJar("c:/wordcount.jar");
		job.setJarByClass(WordCountDriver.class);
		
		//告诉框架，我们的程序所用的mapper类和reducer类
		job.setMapperClass(WordCountMapper.class);
		job.setReducerClass(WordCountReducer.class);
		//使用tCombiner,使用Combiner的好处是让数据在mapper task中就做统计求和，然后将求和后的结果传递给
		//reducer，然后reducer可以在进行求和。这样的好处是减少了reducer的工作。让每个mapper task自己做聚合，
		//通过分担的方式让效率得以提升,由于combiner的内容结构，编程规范也是集成reducer，所以在当前场景中可以将combiner直接
		//设置成WordCountReducer
		job.setCombinerClass(WordCountReducer.class);
		
		//告诉框架，我们的mapperreducer输出的数据类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		// 告诉框架，我们的数据读取、结果输出所用的format组件
		// TextInputFormat是mapreduce框架中内置的一种读取文本文件的输入组件
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		
		// 告诉框架，我们要处理的文件在哪个路径下,注意若hdfs中已经有了/wordcount/input/这个文件，说明
		FileInputFormat.setInputPaths(job, new Path("e:/wordcount/input/"));
		
		// 告诉框架，我们的处理结果要输出到哪里去
		FileOutputFormat.setOutputPath(job, new Path("e:/wordcount/output/"));
		
		boolean res = job.waitForCompletion(true);
		
		System.exit(res?0:1);
	}
}

准备工作：

在E盘下准备e:/wordcount/input/a.txt，其中的内容如下：

The true
nobility is 
in being 
superior to 
your previous 
self guess
No great 
discovery 
was ever 
made without 
a bold 
Knowledge will 
give you 
power but 
character respect
The sun 
is just 
rising in 
the morning 
of another 
day I
I figure 
life is 
a gift 
and I 
don't intend 
on wasting

右键运行上面的代码，进入：

E:\wordcount\output\part-r-00000中看结果，结果内容如下：

I	3
Knowledge	1
No	1
The	2
a	2
and	1
another	1
being	1
bold	1
but	1
character	1
day	1
discovery	1
don't	1
ever	1
figure	1
gift	1
give	1
great	1
guess	1
in	2
intend	1
is	3
just	1
life	1
made	1
morning	1
nobility	1
of	1
on	1
power	1
previous	1
respect	1
rising	1
self	1
sun	1
superior	1
the	1
to	1
true	1
was	1
wasting	1
will	1
without	1
you	1
your	1

经过上面的所有步骤之后，程序已经编写完成


总结：

3.MAPREDUCE中的Combiner[dht1] 

（1）combiner是MR程序中Mapper和Reducer之外的一种组件

（2）combiner组件的父类就是Reducer

（3）combiner和reducer的区别在于运行的位置：

Combiner是在每一个maptask所在的节点运行

Reducer是接收全局所有Mapper的输出结果；

(4) combiner的意义就是对每一个maptask的输出进行局部汇总，以减小网络传输量

具体实现步骤：

    1、 自定义一个combiner继承Reducer，重写reduce方法

    2、 在job中设置： job.setCombinerClass(CustomCombiner.class)

(5) combiner能够应用的前提是不能影响最终的业务逻辑

而且，combiner的输出kv应该跟reducer的输入kv类型要对应起来

Combiner的使用要非常谨慎

因为combiner在mapreduce过程中可能调用也肯能不调用，可能调一次也可能调多次

所以：combiner使用的原则是：有或没有都不能影响业务逻辑


===============================================================================


 流量统计和自定义类实现序列化案例：

package cn.toto.bigdata.mr.wc.flowsum;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.WritableComparable;

/**
 * 自定义的类要被mapreduce使用，需要序列化WritableComparable
 */
public class FlowBean implements WritableComparable<FlowBean> {

	private String phoneNbr;
	private long upFlow;
	private long dFlow;
	private long sumFlow;
	
	/**
	 * 
	 */
	public FlowBean() {}
	
	/**
	 * 序列化框架在反序列化操作创建对象实例时会调用无参构造
	 */
	@Override
	public void write(DataOutput out) throws IOException {
		out.writeUTF(phoneNbr);
		out.writeLong(upFlow);
		out.writeLong(dFlow);
		out.writeLong(sumFlow);
	}

	/**
	 * 反序列化方法，注意：字段的反序列化顺序与序列化时的顺序保持一致
	 */
	@Override
	public void readFields(DataInput in) throws IOException {
		this.phoneNbr = in.readUTF();
		this.upFlow = in.readLong();
		this.dFlow = in.readLong();
		this.sumFlow = in.readLong();
	}

	@Override
	public int compareTo(FlowBean o) {
		return (int)(o.getSumFlow() - this.sumFlow);
	}

	public void set(long upFlow,long dFlow) {
		this.upFlow = upFlow;
		this.dFlow = dFlow;
		this.sumFlow = upFlow + dFlow;
	}
	

	public void set(String phoneNbr,long upFlow, long dFlow) {
		this.phoneNbr = phoneNbr;
		this.upFlow = upFlow;
		this.dFlow = dFlow;
		this.sumFlow = upFlow + dFlow;
	}
	
	public long getUpFlow() {
		return upFlow;
	}

	public void setUpFlow(long upFlow) {
		this.upFlow = upFlow;
	}

	public long getdFlow() {
		return dFlow;
	}

	public void setdFlow(long dFlow) {
		this.dFlow = dFlow;
	}

	public long getSumFlow() {
		return sumFlow;
	}

	public void setSumFlow(long sumFlow) {
		this.sumFlow = sumFlow;
	}
	
	public String getPhoneNbr() {
		return phoneNbr;
	}

	public void setPhoneNbr(String phoneNbr) {
		this.phoneNbr = phoneNbr;
	}

	@Override
	public String toString() {
		return "FlowBean [phoneNbr=" + phoneNbr + ", upFlow=" + upFlow + ", dFlow=" + dFlow + ", sumFlow=" + sumFlow
				+ "]";
	}
}


package cn.toto.bigdata.mr.wc.flowsum;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
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.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class FlowSum {
	
	//在kv中传输我们自定义的对象时可以的，但是必须实现hadoop的序列化机制 implements Writable
	public static class FlowSumMapper extends Mapper<LongWritable, Text, Text, FlowBean> {
		Text k = new Text();
		FlowBean v = new FlowBean();
		
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			//将读到的一行数据进行字段切分
			String line = value.toString();
			String[] fields = StringUtils.split(line,"\t");
			
			//抽取业务所需要的个字段
			String phoneNbr = fields[1];
			long upFlow = Long.parseLong(fields[fields.length - 3]);
			long dFlow = Long.parseLong(fields[fields.length - 2]);
			
			k.set(phoneNbr);
			v.set(upFlow, dFlow);
			
			context.write(k, v);
		}
	}
	
	public static class FlowSumReducer extends Reducer<Text, FlowBean, Text, FlowBean> {
		FlowBean v = new FlowBean();
		
		/**
		 * reduce方法接收到的key是某一组<a手机号，bean><a手机号，bean><a手机号，bean>中的第一个手机号
		 * reduce方法接收到的vlaues是这一组kv中的所有bean的一个迭代器
		 */
		@Override
		protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
			long upFlowCount = 0;
			long dFlowCount = 0;
			for (FlowBean bean : values) {
				upFlowCount += bean.getUpFlow();
				dFlowCount += bean.getdFlow();
			}

			v.set(upFlowCount, dFlowCount);
			context.write(key, v);
		}
	}
	
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();

		/*
		 * conf.set("mapreduce.framework.name", "yarn");
		 * conf.set("yarn.resourcemanager.hostname", "mini1");
		 */

		Job job = Job.getInstance(conf);
		// 告诉框架，我们的程序所在jar包的路径
		// job.setJar("c:/wordcount.jar");
		job.setJarByClass(FlowSum.class);

		// 告诉框架，我们的程序所用的mapper类和reducer类
		job.setMapperClass(FlowSumMapper.class);
		job.setReducerClass(FlowSumReducer.class);

		// 告诉框架，我们的mapperreducer输出的数据类型
		/*
		 * job.setMapOutputKeyClass(Text.class);
		 * job.setMapOutputValueClass(FlowBean.class);
		 */
		// 如果map阶段输出的数据类型跟最终输出的数据类型一致，就只要以下两行代码来指定
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(FlowBean.class);

		// 框架中默认的输入输出组件就是这俩货，所以可以省略这两行代码
		/*
		 * job.setInputFormatClass(TextInputFormat.class);
		 * job.setOutputFormatClass(TextOutputFormat.class);
		 */

		// 告诉框架，我们要处理的文件在哪个路径下
		FileInputFormat.setInputPaths(job, new Path("E:/learnTempFolder/flow/input/"));

		// 告诉框架，我们的处理结果要输出到哪里去
		FileOutputFormat.setOutputPath(job, new Path("E:/learnTempFolder/flow/output/"));

		boolean res = job.waitForCompletion(true);

		System.exit(res ? 0 : 1);
	}
}
package cn.toto.bigdata.mr.wc.flowsum;

import java.io.IOException;

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.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 实现流量汇总并且按照流量大小倒序排序 前提：处理的数据是已经汇总过的结果文件
 * 
 * @author
 * 
 */
public class FlowSumSort {
	
	public static class FlowSumSortMapper extends Mapper<LongWritable, Text, FlowBean, Text> {
		FlowBean k = new FlowBean();
		Text v = new Text();
		
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			String[] fields = line.split("\t");
			String phoneNbr = fields[0];
			long upFlowSum = Long.parseLong(fields[1]);
			long dFlowSum = Long.parseLong(fields[2]);
			k.set(upFlowSum, dFlowSum);
			v.set(phoneNbr);
			context.write(k, v);
		}
	}
	
	public static class FlowSumSortReducer extends Reducer<FlowBean, Text, Text, FlowBean> {

		@Override
		protected void reduce(FlowBean bean, Iterable<Text> phoneNbrs, Context context) throws IOException, InterruptedException {

			context.write(phoneNbrs.iterator().next(), bean);

		}

	}
	
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();

		Job job = Job.getInstance(conf);
		job.setJarByClass(FlowSumSort.class);

		// 告诉框架，我们的程序所用的mapper类和reducer类
		job.setMapperClass(FlowSumSortMapper.class);
		job.setReducerClass(FlowSumSortReducer.class);

		job.setMapOutputKeyClass(FlowBean.class);
		job.setMapOutputValueClass(Text.class);

		// 告诉框架，我们的mapperreducer输出的数据类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(FlowBean.class);

		// 告诉框架，我们要处理的文件在哪个路径下(注意：这里的程序执行)
		FileInputFormat.setInputPaths(job, new Path("E:/learnTempFolder/flow/output/"));

		// 告诉框架，我们的处理结果要输出到哪里去
		FileOutputFormat.setOutputPath(job, new Path("E:/learnTempFolder/flow/sortout/"));

		boolean res = job.waitForCompletion(true);

		System.exit(res ? 0 : 1);
	}
}

运行条件模拟：

1、配置环境变量为HADOOP_HOME=E:\learnTempFolder\hadoop-2.7.3

 2、从CSDN资源上下载支持win10版本的：E:\learnTempFolder\hadoop-2.7.3\bin\winutils.exe 和 E:\learnTempFolder\hadoop-2.7.3\bin\hadoop.dll

界面效果如下：


3、准备要处理的资料：


HTTP_20130313143750.dat 数据文件的具体内容如：

1363157985066 	13726230503	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com		24	27	2481	24681	200
1363157995052 	13826544101	5C-0E-8B-C7-F1-E0:CMCC	120.197.40.4			4	0	264	0	200
1363157991076 	13926435656	20-10-7A-28-CC-0A:CMCC	120.196.100.99			2	4	132	1512	200
1363154400022 	13926251106	5C-0E-8B-8B-B1-50:CMCC	120.197.40.4			4	0	240	0	200
1363157993044 	18211575961	94-71-AC-CD-E6-18:CMCC-EASY	120.196.100.99	iface.qiyi.com	视频网站	15	12	1527	2106	200
1363157995074 	84138413	5C-0E-8B-8C-E8-20:7DaysInn	120.197.40.4	122.72.52.12		20	16	4116	1432	200
1363157993055 	13560439658	C4-17-FE-BA-DE-D9:CMCC	120.196.100.99			18	15	1116	954	200
1363157995033 	15920133257	5C-0E-8B-C7-BA-20:CMCC	120.197.40.4	sug.so.360.cn	信息安全	20	20	3156	2936	200
1363157983019 	13719199419	68-A1-B7-03-07-B1:CMCC-EASY	120.196.100.82			4	0	240	0	200
1363157984041 	13660577991	5C-0E-8B-92-5C-20:CMCC-EASY	120.197.40.4	s19.cnzz.com	站点统计	24	9	6960	690	200
1363157973098 	15013685858	5C-0E-8B-C7-F7-90:CMCC	120.197.40.4	rank.ie.sogou.com	搜索引擎	28	27	3659	3538	200
1363157986029 	15989002119	E8-99-C4-4E-93-E0:CMCC-EASY	120.196.100.99	www.umeng.com	站点统计	3	3	1938	180	200
1363157992093 	13560439658	C4-17-FE-BA-DE-D9:CMCC	120.196.100.99			15	9	918	4938	200
1363157986041 	13480253104	5C-0E-8B-C7-FC-80:CMCC-EASY	120.197.40.4			3	3	180	180	200
1363157984040 	13602846565	5C-0E-8B-8B-B6-00:CMCC	120.197.40.4	2052.flash2-http.qq.com	综合门户	15	12	1938	2910	200
1363157995093 	13922314466	00-FD-07-A2-EC-BA:CMCC	120.196.100.82	img.qfc.cn		12	12	3008	3720	200
1363157982040 	13502468823	5C-0A-5B-6A-0B-D4:CMCC-EASY	120.196.100.99	y0.ifengimg.com	综合门户	57	102	7335	110349	200
1363157986072 	18320173382	84-25-DB-4F-10-1A:CMCC-EASY	120.196.100.99	input.shouji.sogou.com	搜索引擎	21	18	9531	2412	200
1363157990043 	13925057413	00-1F-64-E1-E6-9A:CMCC	120.196.100.55	t3.baidu.com	搜索引擎	69	63	11058	48243	200
1363157988072 	13760778710	00-FD-07-A4-7B-08:CMCC	120.196.100.82			2	2	120	120	200
1363157985066 	13726238888	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com		24	27	2481	24681	200
1363157993055 	13560436666	C4-17-FE-BA-DE-D9:CMCC	120.196.100.99			18	15	1116	954	200
4、先运行FlowSum（右键执行Java程序）

运行生成的文件为E:\learnTempFolder\flow\output\part-r-00000，内容如下：

13480253104	180	180	360
13502468823	7335	110349	117684
13560436666	1116	954	2070
13560439658	2034	5892	7926
13602846565	1938	2910	4848
13660577991	6960	690	7650
13719199419	240	0	240
13726230503	2481	24681	27162
13726238888	2481	24681	27162
13760778710	120	120	240
13826544101	264	0	264
13922314466	3008	3720	6728
13925057413	11058	48243	59301
13926251106	240	0	240
13926435656	132	1512	1644
15013685858	3659	3538	7197
15920133257	3156	2936	6092
15989002119	1938	180	2118
18211575961	1527	2106	3633
18320173382	9531	2412	11943
84138413	4116	1432	5548
5、运行FlowSumSort（注意不要删除上面的part-r-00000）

运行后产生的文件内容是：

13502468823	7335	110349	117684
13925057413	11058	48243	59301
13726230503	2481	24681	27162
18320173382	9531	2412	11943
13560439658	2034	5892	7926
13660577991	6960	690	7650
15013685858	3659	3538	7197
13922314466	3008	3720	6728
15920133257	3156	2936	6092
84138413	4116	1432	5548
13602846565	1938	2910	4848
18211575961	1527	2106	3633
15989002119	1938	180	2118
13560436666	1116	954	2070
13926435656	132	1512	1644
13480253104	180	180	360
13826544101	264	0	264
13719199419	240	0	240

当然，我们也可以一次性求和并运算出结果输出到指定的文件目录中，代码如下：

package cn.toto.bigdata.mr.wc.flowsum;

import java.io.IOException;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;

import org.apache.commons.lang.StringUtils;
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.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class OneStepSumSort {

	public static class OneStepSumSortMapper extends Mapper<LongWritable, Text, Text, FlowBean> {
		Text k = new Text();
		FlowBean v = new FlowBean();
		
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			//将读到的一行数据进行字段切分
			String line = value.toString();
			String[] fields = StringUtils.split(line,"\t");
			
			//抽取业务所需要的各字段
			String phoneNbr = fields[1];
			long upFlow = Long.parseLong(fields[fields.length - 3]);
			long dFlow = Long.parseLong(fields[fields.length - 2]);
			
			k.set(phoneNbr);
			v.set(phoneNbr, upFlow, dFlow);
			
			context.write(k, v);
		}
	}
	
	public static class OneStepSumSortReducer extends Reducer<Text, FlowBean, Text, FlowBean> {
		TreeMap<FlowBean, Text> treeMap = new TreeMap<FlowBean,Text>();
		
		@Override
		protected void reduce(Text key, Iterable<FlowBean> values, Context context)
				throws IOException, InterruptedException {
			int upCount = 0;
			int dCount = 0;
			for (FlowBean bean : values) {
				upCount += bean.getUpFlow();
				dCount += bean.getdFlow();
			}
			FlowBean sumBean = new FlowBean();
			sumBean.set(key.toString(), upCount, dCount);
			
			Text text = new Text(key.toString());
			treeMap.put(sumBean, text);
		}
		
		@Override
		protected void cleanup(Reducer<Text, FlowBean, Text, FlowBean>.Context context)
				throws IOException, InterruptedException {
			Set<Entry<FlowBean, Text>> entrySet = treeMap.entrySet();
			for (Entry<FlowBean, Text> ent : entrySet) {
				context.write(ent.getValue(), ent.getKey());
			}
		}
	}
	
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();

		Job job = Job.getInstance(conf);
		job.setJarByClass(OneStepSumSort.class);

		// 告诉框架，我们的程序所用的mapper类和reducer类
		job.setMapperClass(OneStepSumSortMapper.class);
		job.setReducerClass(OneStepSumSortReducer.class);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(FlowBean.class);

		// 告诉框架，我们的mapperreducer输出的数据类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(FlowBean.class);
		
		// 告诉框架，我们要处理的文件在哪个路径下
		FileInputFormat.setInputPaths(job, new Path("E:/flow/input/"));

	    // 告诉框架，我们的处理结果要输出到哪里去
	    FileOutputFormat.setOutputPath(job, new Path("E:/flow/sortout/"));
		
	    boolean res = job.waitForCompletion(true);
		
	    System.exit(res ? 0 : 1);
	}
}

到"E:/flow/sortout/"目录下，查看结果：


即：

13502468823	FlowBean [phoneNbr=13502468823, upFlow=7335, dFlow=110349, sumFlow=117684]
13925057413	FlowBean [phoneNbr=13925057413, upFlow=11058, dFlow=48243, sumFlow=59301]
13726238888	FlowBean [phoneNbr=13726230503, upFlow=2481, dFlow=24681, sumFlow=27162]
18320173382	FlowBean [phoneNbr=18320173382, upFlow=9531, dFlow=2412, sumFlow=11943]
13560439658	FlowBean [phoneNbr=13560439658, upFlow=2034, dFlow=5892, sumFlow=7926]
13660577991	FlowBean [phoneNbr=13660577991, upFlow=6960, dFlow=690, sumFlow=7650]
15013685858	FlowBean [phoneNbr=15013685858, upFlow=3659, dFlow=3538, sumFlow=7197]
13922314466	FlowBean [phoneNbr=13922314466, upFlow=3008, dFlow=3720, sumFlow=6728]
15920133257	FlowBean [phoneNbr=15920133257, upFlow=3156, dFlow=2936, sumFlow=6092]
84138413	FlowBean [phoneNbr=84138413, upFlow=4116, dFlow=1432, sumFlow=5548]
13602846565	FlowBean [phoneNbr=13602846565, upFlow=1938, dFlow=2910, sumFlow=4848]
18211575961	FlowBean [phoneNbr=18211575961, upFlow=1527, dFlow=2106, sumFlow=3633]
15989002119	FlowBean [phoneNbr=15989002119, upFlow=1938, dFlow=180, sumFlow=2118]
13560436666	FlowBean [phoneNbr=13560436666, upFlow=1116, dFlow=954, sumFlow=2070]
13926435656	FlowBean [phoneNbr=13926435656, upFlow=132, dFlow=1512, sumFlow=1644]
13480253104	FlowBean [phoneNbr=13480253104, upFlow=180, dFlow=180, sumFlow=360]
13826544101	FlowBean [phoneNbr=13826544101, upFlow=264, dFlow=0, sumFlow=264]
13926251106	FlowBean [phoneNbr=13719199419, upFlow=240, dFlow=0, sumFlow=240]

6、为不同的手机号设置分区，让不同的手机号在不同的文件中。方法如下：

A:下面是自定义分区，自定分区的代码如下：

package cn.toto.bigdata.mr.wc.flowsum;

import java.util.HashMap;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

/**
 * 自定义分区要继承Partitioner
 */
public class ProvincePartitioner extends Partitioner<Text, FlowBean>{
	private static HashMap<String, Integer> provinceMap = new HashMap<String, Integer>();
	
	static {
		provinceMap.put("138", 0);
		provinceMap.put("139", 1);
		provinceMap.put("136", 2);
		provinceMap.put("137", 3);
		provinceMap.put("135", 4);
	}
	
	@Override
	public int getPartition(Text key, FlowBean value, int numPartitions) {
		Integer code = provinceMap.get(key.toString().substring(0,3));
		if (code != null) {
			return code;
		}
		return 5;
	}
}
B:测试一下自定义分区： 

package cn.toto.bigdata.mr.wc.flowsum;

import java.io.IOException;

import org.apache.commons.lang.StringEscapeUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class FlowSumProvince {

	public static class FlowSumProvinceMapper extends Mapper<LongWritable, Text, Text, FlowBean> {
		Text k = new Text();
		FlowBean v = new FlowBean();

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

			// 将读到的一行数据进行字段切分
			String line = value.toString();
			String[] fields = StringUtils.split(line, "\t");

			// 抽取业务所需要的各字段
			String phoneNbr = fields[1];
			long upFlow = Long.parseLong(fields[fields.length - 3]);
			long dFlow = Long.parseLong(fields[fields.length - 2]);

			k.set(phoneNbr);
			v.set(phoneNbr, upFlow, dFlow);

			context.write(k, v);

		}
	}
	
	public static class FlowSumProvinceReducer extends Reducer<Text, FlowBean, Text, FlowBean> {
		
		@Override
		protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {

			int upCount = 0;
			int dCount = 0;
			for (FlowBean bean : values) {
				upCount += bean.getUpFlow();
				dCount += bean.getdFlow();

			}
			FlowBean sumBean = new FlowBean();
			sumBean.set(key.toString(), upCount, dCount);
			context.write(key, sumBean);

		}
	}
	
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();

		Job job = Job.getInstance(conf);
		job.setJarByClass(FlowSumProvince.class);

		// 告诉框架，我们的程序所用的mapper类和reducer类
		job.setMapperClass(FlowSumProvinceMapper.class);
		job.setReducerClass(FlowSumProvinceReducer.class);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(FlowBean.class);

		// 告诉框架，我们的mapperreducer输出的数据类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(FlowBean.class);
		
		//设置shuffle的分区组件使用我们自定义的分区组件,按照手机号进行分区,注意在自定义的手机号分区中有5个，所以我们的分区不能少于6个
		job.setPartitionerClass(ProvincePartitioner.class);
		
		//设置reduce task的数量
		job.setNumReduceTasks(6);
		
		//告诉框架，我们要处理的文件在哪个路径下
		FileInputFormat.setInputPaths(job, new Path("E:/flow/input/"));
		
		//告诉框架，我们的处理结果要输出到哪里去
		Path out = new Path("E:/flow/provinceout/");
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(out)) {
			fs.delete(out,true);
		}
		
		FileOutputFormat.setOutputPath(job, out);
		boolean res = job.waitForCompletion(true);
		System.exit(res ? 0 : 1);
	}
}
C:运行所需的准备：

数据文件:


文件内容如下：

1363157985066 	13726230503	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com		24	27	2481	24681	200
1363157995052 	13826544101	5C-0E-8B-C7-F1-E0:CMCC	120.197.40.4			4	0	264	0	200
1363157991076 	13926435656	20-10-7A-28-CC-0A:CMCC	120.196.100.99			2	4	132	1512	200
1363154400022 	13926251106	5C-0E-8B-8B-B1-50:CMCC	120.197.40.4			4	0	240	0	200
1363157993044 	18211575961	94-71-AC-CD-E6-18:CMCC-EASY	120.196.100.99	iface.qiyi.com	视频网站	15	12	1527	2106	200
1363157995074 	84138413	5C-0E-8B-8C-E8-20:7DaysInn	120.197.40.4	122.72.52.12		20	16	4116	1432	200
1363157993055 	13560439658	C4-17-FE-BA-DE-D9:CMCC	120.196.100.99			18	15	1116	954	200
1363157995033 	15920133257	5C-0E-8B-C7-BA-20:CMCC	120.197.40.4	sug.so.360.cn	信息安全	20	20	3156	2936	200
1363157983019 	13719199419	68-A1-B7-03-07-B1:CMCC-EASY	120.196.100.82			4	0	240	0	200
1363157984041 	13660577991	5C-0E-8B-92-5C-20:CMCC-EASY	120.197.40.4	s19.cnzz.com	站点统计	24	9	6960	690	200
1363157973098 	15013685858	5C-0E-8B-C7-F7-90:CMCC	120.197.40.4	rank.ie.sogou.com	搜索引擎	28	27	3659	3538	200
1363157986029 	15989002119	E8-99-C4-4E-93-E0:CMCC-EASY	120.196.100.99	www.umeng.com	站点统计	3	3	1938	180	200
1363157992093 	13560439658	C4-17-FE-BA-DE-D9:CMCC	120.196.100.99			15	9	918	4938	200
1363157986041 	13480253104	5C-0E-8B-C7-FC-80:CMCC-EASY	120.197.40.4			3	3	180	180	200
1363157984040 	13602846565	5C-0E-8B-8B-B6-00:CMCC	120.197.40.4	2052.flash2-http.qq.com	综合门户	15	12	1938	2910	200
1363157995093 	13922314466	00-FD-07-A2-EC-BA:CMCC	120.196.100.82	img.qfc.cn		12	12	3008	3720	200
1363157982040 	13502468823	5C-0A-5B-6A-0B-D4:CMCC-EASY	120.196.100.99	y0.ifengimg.com	综合门户	57	102	7335	110349	200
1363157986072 	18320173382	84-25-DB-4F-10-1A:CMCC-EASY	120.196.100.99	input.shouji.sogou.com	搜索引擎	21	18	9531	2412	200
1363157990043 	13925057413	00-1F-64-E1-E6-9A:CMCC	120.196.100.55	t3.baidu.com	搜索引擎	69	63	11058	48243	200
1363157988072 	13760778710	00-FD-07-A4-7B-08:CMCC	120.196.100.82			2	2	120	120	200
1363157985066 	13726238888	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com		24	27	2481	24681	200
1363157993055 	13560436666	C4-17-FE-BA-DE-D9:CMCC	120.196.100.99			18	15	1116	954	200

运行后的结果如下：



part-r-00001中内容：




等等






 

上一篇 下一篇
优秀的个人博客,低调大师

微信关注我们

原文链接：https://yq.aliyun.com/articles/100924

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

MapReduce原理与设计思想（转载：http://blog.jobbole.com/80619/）

简单解释 MapReduce 算法 一个有趣的例子 你想数出一摞牌中有多少张黑桃。直观方式是一张一张检查并且数出有多少张是黑桃 MapReduce方法则是 给在座的所有玩家中分配这摞牌 让每个玩家数自己手中的牌有几张是黑桃然后把这个数目汇报给你 你把所有玩家告诉你的数字加起来得到最后的结论 拆分 MapReduce合并了两种经典函数 映射Mapping对集合里的每个目标应用同一个操作。即如果你想把表单里每个单元格乘以二那么把这个函数单独地应用在每个单元格上的操作就属于mapping。 化简Reducing 遍历集合中的元素来返回一个综合的结果。即输出表单里一列数字的和这个任务属于reducing。 重新审视上面的例子 重新审视我们原来那个分散纸牌的例子我们有MapReduce数据分析的基本方法。友情提示这不是个严谨的例子。在这个例子里人代表计算机因为他们同时工作所以他们是个集群。在大多数实际应用中我们假设数据已经在每台计算机上了 – 也就是说把牌分发出去并不是MapReduce的一步。事实上在计算机集群中如何存储文件是Hadoop的真正核心。 通过把牌分给多个玩家并且让他们各自数数你就...
2017-06-02
534
上一篇

ElasticSearch 5.3.3 和 5.4.1 发布

ElasticSearch 5.3.3 和 5.4.1 发布了，Elastic Search 是一个基于 Lucene 构建的开源，分布式，RESTful 搜索引擎。设计用于云计算中，能够达到实时搜索，稳定，可靠，快速，安装使用方便。支持通过 HTTP 使用 JSON 进行数据索引。更新如下： 5.3.3 Enhancements Network Disable the Netty recycler in the client #24793 (issues: #22452, #24721) Bug fixes Index APIs Fix legacy GeoPointField decoding in FieldStats #24534 (issue: #24275) Packaging Handle parentheses in batch file path #24731 (issue: #24712) Search Fix RangeFieldMapper rangeQuery to properly handle relations #24808 (issue: #24744...
2017-06-02
629
下一篇

相关文章

发表评论

资源下载

更多资源
优质分享App

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

时间: 2025-12-03 大小: 4.36MB 下载: 23次
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

时间: 2025-12-12 大小: 211.28KB 下载: 54次
Spring

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

时间: 2025-12-17 大小: 5MB 下载: 1次
Sublime Text

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。

时间: 2025-12-03 大小: 9.84MB 下载: 39次
手机查看

扫码在手机上查看文章

扫描二维码，手机阅读更方便

热门标签
OpenAi Nacos3 Service Mesh Gemini3 Rocky Docker DeepSeek HarmonyOS6 Redis SpringCloud Kubernetes SpringBoot4 Jdk25

欢迎您来访！

登录后，您将获得更多功能！
热门文章
联系我们

有任何问题或合作意向欢迎联系我们

Email: 99873273@qq.com

QQ: 99873273

用户登录
用户注册