Hadoop下MapReduce實現Pi值的計算

Hadoop自帶的例子中，有一個計算Pi值的例子。

 

這個程序的原理是這樣的。假如有一個邊長為1的正方形。以正方形的一個端點為圓心，以1為半徑，畫一個圓弧，於是在正方形內就有了一個直角扇形。在正方形里隨機生成若干的點，則有些點是在扇形內，有些點是在扇形外。正方形的面積是1，扇形的面積是0.25*Pi。設點的數量一共是n，扇形內的點數量是nc，在點足夠多足夠密集的情況下，會近似有nc/n的比值約等於扇形面積與正方形面積的比值，也就是nc/n= 0.25*Pi/1，即Pi = 4*nc/n。

 

在正方形內生成的樣本點越多，計算Pi值越精確，這樣，這個問題就很適合用Hadoop來處理啦。假設要在正方形內生成1000萬個點，可以設置10個Map任務，每個Map任務處理100萬個點，也可以設置100個Map任務，每個Map任務處理10萬個點。

 

package mapreduce1;
/*
 *  @create by 劉大哥
 *  2019年9月3日
   *       利用MapReduce計算pi值
 * */
import java.io.IOException;  
import java.util.StringTokenizer;  
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.Mapper;  
import org.apache.hadoop.mapreduce.Reducer;  
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import PI.Pi;  


    public class WordCount {  
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {  
            Job job = Job.getInstance();  
            job.setJobName("WordCount");  
            job.setJarByClass(WordCount.class);  
            job.setMapperClass(doMapper.class);
            job.setReducerClass(doReducer.class);  
            job.setOutputKeyClass(Text.class);  
            job.setOutputValueClass(IntWritable.class);  
            Path in = new Path("hdfs://192.168.100.129:9000/user/hadoop/p1i.txt");   //輸入路徑
            Path out = new Path("hdfs://192.168.100.129:9000/user/hadoop/out_pi1");  //輸出路徑
            FileInputFormat.addInputPath(job, in);  
            FileOutputFormat.setOutputPath(job, out);  
            System.exit(job.waitForCompletion(true) ? 0 : 1);  
        }  
        public static class doMapper extends Mapper<Object, Text, Text, IntWritable>{  
            private static final IntWritable one = new IntWritable(1);  
            @Override  
            protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {  
                String line =  value.toString();    
                String word = line.toString();         //讀取每個map的數值
                //System.out.println(word);
                int num = Integer.parseInt(word);      //轉化為int類型 
                //System.out.println(num);
                int[] base = {2,5};
                Pi test = new Pi(base);  
                int a= 0;         // 是否在扇形區域內的標志符  1：在扇形區域內 2：不在扇形區域內
                int count = 0;  // 統計在扇形區域內點的個數
                for(int x = 0; x < num; x++){
                    double[] t = test.getNext();
                    if(t[0]*t[0]+t[1]*t[1]<1) {  //在扇形區域內
                        a=1;
                        count++;                 //在扇形區域內的個數加+
                    }
                    else {                       //不在扇形區域內
                        a=2;
                    }
                
                }
                double result= count*4.00000000/num;    //每個map計算出pi的值
                String strresule = String.valueOf(result); 
                Text textresult = new Text();              /*轉換類型為Text */
                textresult.set(strresule);                
                context.write(textresult, one);         //寫入
            }  
        }  
        public static class doReducer extends Reducer<Text, IntWritable, Text, IntWritable>{   //reduce 整合輸出
            private IntWritable result = new IntWritable();  
            @Override  
            protected void reduce(Text key, Iterable<IntWritable> values, Context context)  
            throws IOException, InterruptedException {  
            int sum = 0;  
            for (IntWritable value : values) {  
                sum += value.get();  
            }  
                result.set(sum); 
                context.write(key, result);  
            }  
        }  
    }