利用MapReduce计算平均数

Posted 2020-11-05 qwangxiao

利用mapreduce求出股票价格的开盘和收盘平均数

下图为采集到的股票信息，共计1416支股票的信息

因为在linux系统下默认采用utf-8的编码格式，而在win下txt默认采用ANSI编码格式。所以需要在linux下将文件转换一下格式，可以采用：

递归转换(包括子文件夹)

find default -type d -exec mkdir -p utf/{} \;

find default -type f -exec iconv -f GBK -t UTF-8 {} -o utf/{} \;

 

这两行命令将default目录下的文件由GBK编码转换为UTF-8编码，目录结构不变，转码后的文件保存在utf/default目录下。

[java] view plain copy

 

1. package economic;  
2.   
3. import java.io.IOException;  
4. import java.util.Iterator;  
5. import java.util.StringTokenizer;  
6.   
7. import org.apache.hadoop.conf.Configuration;  
8. import org.apache.hadoop.fs.Path;  
9. import org.apache.hadoop.io.FloatWritable;  
10. import org.apache.hadoop.io.IntWritable;  
11. import org.apache.hadoop.io.LongWritable;  
12. import org.apache.hadoop.io.Text;  
13. import org.apache.hadoop.mapreduce.Job;  
14. import org.apache.hadoop.mapreduce.Mapper;  
15. import org.apache.hadoop.mapreduce.Reducer;  
16. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
17. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
18. import org.apache.hadoop.util.GenericOptionsParser;  
19.   
20. public class ScoreAvgTest {  
21.     /** 
22.      *  
23.      * @author hadoop www.ysgj1688.com KEYIN：输入map的key值，为每行文本的开始位置子字节计算，（0,11...） 
24.      *         VALUEIN：输入map的value，为每行文本值 KEYOUT ：输出的key值 VALUEOUT：输出的value值 
25.      */  
26.     public static class MapperClass extends Mapper<Object, Text, Text, Text> {  
27.         private Text companyName = new Text();  
28.         private Text open = new Text();  
29.           
30.         private Text data=new Text();  
31.         private int n = 0;  
32.   
33.         @Override  
34.         protected void map(Object key, Text value, Context context)  
35.                 throws IOException, InterruptedException {  
36.             // TODO Auto-generated method stub  
37.   
38.             System.out.println(this.n);  
39.             n++;  
40.   
41.             String lineText = value.toString();  
42.             String[] args = lineText.split("\\s+");  
43.   
44.             if (args.length == 4) {  
45.                 this.companyName.set(args[1]);  
46.             }  
47.             if (args.length == 7) {  
48.                 try {  
49.                     System.out.println("Bfter Reducer:" + companyName + ","  
50.                             + args[1]);  
51.                     data.set(args[1]+" "+args[4]);  
52.                     context.write(this.companyName, data);  
53.                 } catch (IOException e) {  
54.                     e.printStackTrace(www.vboyl130.cn);  
55.                 } catch (InterruptedException e) {  
56.                     e.printStackTrace();  
57.                 }  
58.             }  
59.   
60.         }  
61.   
62.     }  
63.   
64.     /** 
65.      *  
66.      * @author hadoop KEYIN:输入的名字 VALUEIN：输入的分数 KEYOUT：输出的名字 VALUEOUT：统计输出的平均分 
67.      */  
68.     public static class ReducerClass extends Reducer<Text, Text, Text, Text> {  
69.   
70.         private Text text = new Text();  
71.   
72.         protected void reduce(Text companyName, Iterable<Text> kaipan,  
73.                 Context context) throws IOException, InterruptedException {  
74.             // TODO Auto-generated method stub  
75.   
76.             double sumOpen = 0.0;  
77.             double sumClose = 0.0;  
78.             int num = 0;  
79.   
80.             Iterator<Text> $it = kaipan.iterator();  
81.             while ($it.hasNext())www.vboyule.cn  {  
82.                 String record = $it.next(www.baohuayule.net ).toString();  
83.                 String[] getData=record.split(" ");  
84.                 System.out.println(num);  
85.                 System.out.println("原始数据：" + record);  
86.                 num++;  
87.                 System.out.println("第" www.qinlinyule.cn + num + "次循环");  
88.                 sumOpen += (Double.valueOf(getData[0])*100);  
89.                 sumClose+=(Double.valueOf(getData[1])*100);  
90.             }  
91.             double openPrise = sumOpen / (100 * num);  
92.             double closePrise = sumClose / (100 * num);  
93.             System.out.println("openPrice1:" + openPrise);  
94.             System.out.println("www.120xh.cn  closePrice1:" + closePrise);  
95.             openPrise = (double) Math.round(openPrise * 100) / 100;  
96.             closePrise = (double) Math.round(closePrise * 100) / 100;  
97.             System.out.println("sumOpen:" + sumOpen+"   sumClose"+sumClose);  
98.             System.out.println("openPrice2:" + openPrise);  
99.             System.out.println("closePrice2:" + closePrise);  
100.             String result ="开盘平均价："+Double.toString(openPrise)+",   收盘平均价："+Double.toString(closePrise);    
101.             text.set(result);  
102.             try {  
103.                 context.write(companyName, text);  
104.             } catch (IOException e) {  
105.                 e.printStackTrace();  
106.             } catch (InterruptedException e) {  
107.                 e.printStackTrace();  
108.             }  
109.         }  
110.   
111.     }  
112.   
113.     public static void main(String[] args) throws IOException,  
114.             InterruptedException, www.255055.cn/   ClassNotFoundException {  
115.   
116.         Configuration conf = new Configuration();  
117.         conf.set("fs.default.name", "hdfs://localhost:9000");  
118.         String[] otherArgs = new String[] { "export", "output" };    //export为文本输入路径，output为输出路径  
119.         if (otherArgs.length < 2) {  
120.             System.err.println("Usage:wordcount<in>[<in>...]<out>");  
121.             System.exit(2);  
122.         }  
123.         Job job = Job.getInstance(conf, "arg");  
124.         job.setJarByClass(ScoreAvgTest.class);  
125.         job.setMapperClass(MapperClass.class);  
126. //      job.setCombinerClass(ReducerClass.class);  
127.         System.out.println("Mapper over");  
128.         job.setReducerClass(ReducerClass.class);  
129.         System.out.println("Reducer over");  
130.         job.setOutputKeyClass(Text.class);  
131.         job.setOutputValueClass(Text.class);  
132.         FileInputFormat.addInputPath(job, new Path(otherArgs[0]));  
133.         FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));  
134.         System.exit(job.waitForCompletion(true) ? 0 : 1);  
135.   
136.     }  
137. }  

[java] view plain copy

 

1.   

运行后生成的output文件夹中的文件