devbox
IT小白
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

MapReduce基础回顾题目_分别统计各柜台的销售总额,并对各柜台的销售总额进行降序排序。

作者：IT小白 | 2024-06-22 23:21:47

赞

踩

分别统计各柜台的销售总额,并对各柜台的销售总额进行降序排序。
week1,A0001,10,20
week1,A0002,8.5,15
week1,A0003,9.2,30
week1,B0001,10.5,50
week2,A0001,11,30
week2,A0002,8,20
week2,A0003,9.2,20
week2,B0001,10,55
week3,A0001,9.5,10
week3,A0002,8.8,30
week3,A0003,9.8,30
week3,B0001,9,58
week4,A0001,9.2,14
week4,A0002,8.5,22
week4,A0003,10.3,45
week4,B0001,7,12

要求：根据以上数据，用 MapReduce 统计出如下数据：
1、每种商品的销售总金额，并降序排序
2、每种商品销售额最多的三周

问题一：分组求和，总体降序排列。
该问题采用两个mapreduce程序进行解决的，第一个求出每种商品的销售总额，第二个进行降序排序。

  1.  
  2. import java.io.IOException;
  3.  
  4. import org.apache.hadoop.conf.Configuration;
  5. import org.apache.hadoop.fs.FileSystem;
  6. import org.apache.hadoop.fs.Path;
  7. import org.apache.hadoop.io.DoubleWritable;
  8. import org.apache.hadoop.io.LongWritable;
  9. import org.apache.hadoop.io.NullWritable;
  10. import org.apache.hadoop.io.Text;
  11. import org.apache.hadoop.mapreduce.Job;
  12. import org.apache.hadoop.mapreduce.Mapper;
  13. import org.apache.hadoop.mapreduce.Reducer;
  14. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  15. import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob;
  16. import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;
  17. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  18.  
  19.  
  20. /**
  21.  * 1、每种商品的销售总金额，并降序排序
  22.  * 	周数  商品编号  单价  销量（件）
  23.  * 数据：week1	A0001  10    20
  24.  * 2、每种商品销售额最多的三周
  25.  *
  26.  */
  27. public class QuestionMR_1_1 {
  28. 	public static void main(String[] args) throws Exception {
  29. 		Configuration conf = new Configuration();
  30. 		FileSystem fs = FileSystem.get(conf);
  31. 		
  32. 		/**
  33. 		 * 在本地跑把下面两行注释掉
  34. 		 */
  35. 		//conf.set("fs.defaultFS", "hdfs://hadoop01:9000");
  36. 		//System.setProperty("HADOOP_USER_NAME", "hadoop");
  37. 		
  38. 		
  39. 		Job job = Job.getInstance(conf);
  40. 		job.setJarByClass(QuestionMR_1_1.class);
  41. 		
  42. 		job.setMapperClass(MRMapper.class);
  43. 		job.setReducerClass(MRReducer.class);
  44. 		
  45. 		job.setOutputKeyClass(Text.class);
  46. 		job.setOutputValueClass(DoubleWritable.class);
  47. 		
  48. 		/**
  49. 		 * 自定义分区组件
  50. 		 */
  51. 		//job.setPartitionerClass(Object.class);
  52. 		
  53. 		/**
  54. 		 * 自定义分组组件
  55. 		 */
  56. 		//job.setGroupingComparatorClass(Object.class);
  57. 		
  58. 		/**
  59. 		 * 必要的时候使用自定义Combine组件
  60. 		 */
  61. 		//job.setCombinerClass(MRReducer.class);
  62. 		
  63. 	
  64. 		Path inputPath = new Path("G:/test/q1/input");
  65. 		Path outputPath = new Path("G:/test/q1/output_1_1");
  66. 		if(fs.exists(outputPath)){
  67. 			fs.delete(outputPath, true);
  68. 		}
  69. 		
  70. 		FileInputFormat.setInputPaths(job, inputPath);
  71. 		FileOutputFormat.setOutputPath(job, outputPath);
  72. 		
  73. 		
  74. 		
  75. 		Job job2 = Job.getInstance(conf);
  76. 		job2.setJarByClass(QuestionMR_1_1.class);
  77. 		
  78. 		job2.setMapperClass(MRMapper2.class);
  79. 		job2.setReducerClass(MRReducer2.class);
  80. 		
  81. 		job2.setOutputKeyClass(Product.class);
  82. 		job2.setOutputValueClass(NullWritable.class);
  83. 		
  84. 		
  85. 	
  86. 		Path inputPath2 = new Path("G:/test/q1/output_1_1");
  87. 		Path outputPath2 = new Path("G:/test/q1/output_1_2");
  88. 		if(fs.exists(outputPath2)){
  89. 			fs.delete(outputPath2, true);
  90. 		}
  91. 		
  92. 		FileInputFormat.setInputPaths(job2, inputPath2);
  93. 		FileOutputFormat.setOutputPath(job2, outputPath2);
  94. 		//使用连个mapreduce程序进行关联操作
  95. 		JobControl control = new JobControl("CF");
  96. 		
  97. 		ControlledJob conjob1 = new ControlledJob(job.getConfiguration());
  98. 		ControlledJob conjob2 = new ControlledJob(job2.getConfiguration());
  99. 		
  100. 		conjob2.addDependingJob(conjob1);
  101. 		
  102. 		control.addJob(conjob1);
  103. 		control.addJob(conjob2);
  104. 		
  105. 		Thread t = new Thread(control);
  106. 		t.start();
  107. 		
  108. 		while(!control.allFinished()){
  109. 			Thread.sleep(1000);
  110. 		}
  111.  
  112. 		System.exit(0);
  113. 		
  114. 	}
  115. 	
  116. 	//week1	A0001商品  10单价  20数量
  117. 	//每种商品的销售总金额，并降序排序
  118. 	public static class MRMapper extends Mapper<LongWritable, Text, Text, DoubleWritable>{
  119. 		Text k = new Text();
  120. 		DoubleWritable v = new DoubleWritable();
  121. 		
  122. 		@Override
  123. 		protected void map(LongWritable key, Text value,Context context)
  124. 				throws IOException, InterruptedException {
  125. 			
  126. 			String[] line = value.toString().split(",");
  127. 			//根据商品ID作为Key这样在reduce阶段我们就能对每一种商品的销售额进行累加
  128. 			k.set(line[1]);
  129. 			//根据数据格式，求出每条记录的销售总金额
  130. 			double sum = Double.parseDouble(line[2])*Double.parseDouble(line[3]);
  131. 			v.set(sum);
  132. 			context.write(k, v);
  133.  
  134. 		}
  135. 	}
  136. 	
  137. 	public static class MRReducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable>{
  138. 		DoubleWritable v = new DoubleWritable();
  139. 		@Override
  140. 		protected void reduce(Text key, Iterable<DoubleWritable> values, Context context)
  141. 				throws IOException, InterruptedException {
  142. 			
  143. 			double sum = 0D;
  144. 			//相同的商品会在一个reduceTask中进行处理，
  145. 			//这里能够对每一种商品的销售额进行累加
  146. 			for (DoubleWritable num : values) {
  147. 				sum += num.get(); 
  148. 			}
  149. 			v.set(sum);
  150. 			context.write(key,v);
  151. 		}
  152. 	}
  153. 	
  154. 	/**
  155. 	 * 对商品进行排序的时候这里选择使用自定义数据类型的方式进行
  156. 	 * 使用compareTo方法对数据进行排序
  157. 	 */
  158. 	public static class MRMapper2 extends Mapper<LongWritable, Text, Product, NullWritable>{
  159. 		@Override
  160. 		protected void map(LongWritable key, Text value,Context context)
  161. 				throws IOException, InterruptedException {
  162. 			
  163. 			String[] line = value.toString().split("\t");
  164. 			
  165. 			Product p = new Product(line[0], Double.parseDouble(line[1]),"");
  166. 			context.write(p, NullWritable.get());
  167.  
  168. 		}
  169. 	}
  170. 	
  171. 	public static class MRReducer2 extends Reducer<Product, NullWritable, Product, NullWritable>{
  172. 		@Override
  173. 		protected void reduce(Product key, Iterable<NullWritable> values, Context context)
  174. 				throws IOException, InterruptedException {
  175. 			
  176. 			context.write(key, NullWritable.get());
  177. 			
  178. 		}
  179. 	}
  180. }
问题二：每种商品销售额最多的三周是一种简单的分组取TopN
根据商品进行分组，每一组根据销售额进行倒序排序，取Top3即可。
这里使用了自定义分组组件，这样能够适当的减少reduce端的循环次数。

排序通过自定义数据类型Product中的compareTo方法来控制

  1.  
  2. import java.io.IOException;
  3.  
  4. import org.apache.hadoop.conf.Configuration;
  5. import org.apache.hadoop.fs.FileSystem;
  6. import org.apache.hadoop.fs.Path;
  7. import org.apache.hadoop.io.DoubleWritable;
  8. import org.apache.hadoop.io.LongWritable;
  9. import org.apache.hadoop.io.NullWritable;
  10. import org.apache.hadoop.io.Text;
  11. import org.apache.hadoop.mapreduce.Job;
  12. import org.apache.hadoop.mapreduce.Mapper;
  13. import org.apache.hadoop.mapreduce.Reducer;
  14. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  15. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  16.  
  17. /**
  18.  *  数据：week1	A0001  10  20
  19.  *  2、每种商品销售额最多的三周
  20.  * @author Administrator
  21.  *
  22.  */
  23. public class QuestionMR_1_2 {
  24. 	
  25. 	public static void main(String[] args) throws Exception {
  26. 		Configuration conf = new Configuration();
  27. 		FileSystem fs = FileSystem.get(conf);
  28. 		
  29. 		/**
  30. 		 * 在本地跑把下面两行注释掉
  31. 		 */
  32. 		//conf.set("fs.defaultFS", "hdfs://hadoop01:9000");
  33. 		//System.setProperty("HADOOP_USER_NAME", "hadoop");
  34. 		
  35. 		
  36. 		Job job = Job.getInstance(conf);
  37. 		job.setJarByClass(QuestionMR_1_1.class);
  38. 		
  39. 		job.setMapperClass(MRMapper.class);
  40. 		job.setReducerClass(MRReducer.class);
  41. 		
  42. 		job.setOutputKeyClass(Product.class);
  43. 		job.setOutputValueClass(NullWritable.class);
  44. 		
  45. 		/**
  46. 		 * 自定义分区组件
  47. 		 */
  48. 		//job.setPartitionerClass(Object.class);
  49. 		
  50. 		/**
  51. 		 * 自定义分组组件
  52. 		 */
  53. 		job.setGroupingComparatorClass(ProductGC.class);
  54. 		
  55. 		/**
  56. 		 * 必要的时候使用自定义Combine组件
  57. 		 */
  58. 		//job.setCombinerClass(MRReducer.class);
  59. 		
  60. 	
  61. 		Path inputPath = new Path("G:/exam/q1/input");
  62. 		Path outputPath = new Path("G:/exam/q1/output_2_1");
  63. 		if(fs.exists(outputPath)){
  64. 			fs.delete(outputPath, true);
  65. 		}
  66. 		
  67. 		FileInputFormat.setInputPaths(job, inputPath);
  68. 		FileOutputFormat.setOutputPath(job, outputPath);
  69. 		
  70. 		boolean isDone = job.waitForCompletion(true);
  71. 		System.exit(isDone?0:1);
  72. 		
  73. 	}
  74. 	
  75. 	//数据：week1	A0001  10  20
  76. 	public static class MRMapper extends Mapper<LongWritable, Text, Product, NullWritable>{
  77. 		Text k = new Text();
  78. 		DoubleWritable v = new DoubleWritable();
  79. 		//直接将数据读取以后构建自定义数据类型（bean）
  80. 		//然后通过自定义分组组件进行分组
  81. 		//自定义数据类型会根据compareTo方法自动排序
  82. 		@Override
  83. 		protected void map(LongWritable key, Text value,Context context)
  84. 				throws IOException, InterruptedException {
  85. 			
  86. 			String[] line = value.toString().split(",");
  87. 			double sum = Double.parseDouble(line[2])*Double.parseDouble(line[3]);
  88. 			Product p = new Product(line[1],sum,line[0]);
  89. 			context.write(p,NullWritable.get());
  90.  
  91. 		}
  92. 	}
  93. 	
  94. 	public static class MRReducer extends Reducer<Product, NullWritable, Product, NullWritable>{
  95. 		@Override
  96. 		protected void reduce(Product key, Iterable<NullWritable> values, Context context)
  97. 				throws IOException, InterruptedException {
  98. 			
  99. 			int count = 0;
  100. 			for (NullWritable nv : values) {
  101. 				if(count < 3){
  102. 					context.write(key, nv);
  103. 				}
  104. 				count++;
  105. 			}
  106. 		}
  107. 	}
  108. }

自定义分组组件：ProductGC

  1.  
  2. import org.apache.hadoop.io.WritableComparable;
  3. import org.apache.hadoop.io.WritableComparator;
  4.  
  5. public class ProductGC extends WritableComparator{
  6.  
  7. 	public ProductGC() {
  8. 		super(Product.class,true);
  9. 	}
  10.  
  11. 	@Override
  12. 	public int compare(WritableComparable a, WritableComparable b) {
  13. 		Product pa = (Product) a;	
  14. 		Product pb = (Product) b;	
  15. 		
  16. 		//根据商品的ID进行分组
  17. 		return pa.getNamenum().compareTo(pb.getNamenum());
  18. 	}
  19. 	
  20. }

自定义数据类型：Product

  1. package cn.zhao.exam.mapreduce.q1;
  2.  
  3. import java.io.DataInput;
  4. import java.io.DataOutput;
  5. import java.io.IOException;
  6.  
  7. import org.apache.hadoop.io.WritableComparable;
  8.  
  9. public class Product implements WritableComparable<Product>{
  10. 	private String namenum;
  11. 	private double price;
  12. 	private int count;
  13. 	private double totalMoney;
  14. 	private String week;
  15. 	
  16. 	
  17. 	public String getWeek() {
  18. 		return week;
  19. 	}
  20. 	public void setWeek(String week) {
  21. 		this.week = week;
  22. 	}
  23. 	public String getNamenum() {
  24. 		return namenum;
  25. 	}
  26. 	public void setNamenum(String namenum) {
  27. 		this.namenum = namenum;
  28. 	}
  29. 	public double getPrice() {
  30. 		return price;
  31. 	}
  32. 	public void setPrice(double price) {
  33. 		this.price = price;
  34. 	}
  35. 	public int getCount() {
  36. 		return count;
  37. 	}
  38. 	public void setCount(int count) {
  39. 		this.count = count;
  40. 	}
  41. 	public double getTotalMoney() {
  42. 		return totalMoney;
  43. 	}
  44. 	public void setTotalMoney(double totalMoney) {
  45. 		this.totalMoney = totalMoney;
  46. 	}
  47. 	public Product() {
  48. 		super();
  49. 		// TODO Auto-generated constructor stub
  50. 	}
  51. 	
  52. 	
  53. 	
  54. 	public Product(String namenum, double totalMoney, String week) {
  55. 		super();
  56. 		this.namenum = namenum;
  57. 		this.totalMoney = totalMoney;
  58. 		this.week = week;
  59. 	}
  60. 	@Override
  61. 	public String toString() {
  62. 		if("".equals(week)){
  63. 			return  namenum +", "+ totalMoney;
  64. 		}else{
  65. 			return  namenum +", " + week +", " + totalMoney;
  66. 		}
  67. 	}
  68. 	@Override
  69. 	public void write(DataOutput out) throws IOException {
  70. 		out.writeUTF(namenum);
  71. 		out.writeDouble(totalMoney);
  72. 		out.writeUTF(week);
  73. 	}
  74. 	@Override
  75. 	public void readFields(DataInput in) throws IOException {
  76. 		namenum = in.readUTF();
  77. 		totalMoney = in.readDouble();
  78. 		week = in.readUTF();
  79. 		
  80. 	}
  81. 	
  82. 	@Override
  83. 	public int compareTo(Product o) {
  84. 		//这里使用两个字段是因为第二问也使用了这个bean
  85. 		if(o.getNamenum().compareTo(this.getNamenum()) == 0){
  86. 			//控制倒叙排序 o - this > 0 返回1 为降序
  87. 			//this - o > 0 返回 1 为升序
  88. 			double flag = o.totalMoney - this.totalMoney;
  89. 			if(flag == 0){
  90. 				return 0;
  91. 			}else if(flag > 0){
  92. 				return 1;
  93. 			}else{
  94. 				return -1;
  95. 			}
  96. 		}else{
  97. 			return o.getNamenum().compareTo(this.getNamenum());
  98. 		}
  99. 		
  100. 	}
  101. 	
  102. 	
  103. }
本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号