MapReduce编程练习（一）

蒙特卡罗概率算法求π值的MapReduce并行实现

分析：我们可以按如下方法让计算变得可并行化:

1. 在正方形内随机地生成一些点

2. 计算这些点在圆形内的个数

3. p = 圆形内的点的个数

正方形内的点的个数

4. π = 4p

（1）Driver：

public class piDriver {
public static void main(String args[]) throws IOException, InterruptedException, ClassNotFoundException{
getData.data();//生成一组随即坐标点
IHDFSOperator hdfs=new HDFSOperator();
hdfs.putFile("/home/server/list/data.txt", "hdfs://localhost:9000/input/data.txt");//将写有坐标点的文件上传到HDFS
Configuration conf =new Configuration();
Job job=new Job(conf,"Calculate pi");
job.setJarByClass(piDriver.class);
job.setMapperClass(piMap.class);
job.setCombinerClass(piCombiner.class);
job.setReducerClass(piReduce.class);
job.setInputFormatClass(KeyValueInputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path("hdfs://localhost:9000/input/data.txt"));
FileOutputFormat.setOutputPath(job, new Path("hdfs://localhost:9000/output"));
job.waitForCompletion(true);
}

}

（2）Mapper：

public class piMap extends Mapper<Text,Text,Text,IntWritable>{
Text countCircle=new Text();
IntWritable number=new IntWritable();

public void map(Text key,Text value,Context context)
throws IOException, InterruptedException{
int num=0;
String point=value.toString();
String xy[]=point.split(",");
Float x=Float.parseFloat(xy[0]);
System.out.println("x="+x);
Float y=Float.parseFloat(xy[1]);
System.out.println("y="+y);
if(x*x+y*y<=1)//计算所标是否在圆内
num++;
System.out.println("num="+num);
countCircle.set("countCircle");
number.set(num);

context.write(countCircle, number);

}

}

（3）Combiner：

public class piCombiner extends Reducer<Text,IntWritable,Text,IntWritable> {//将一个任务分片的所有map输出做一个汇总
Text temp=new Text();
IntWritable plus=new IntWritable();
public void reduce(Text key,Iterable<IntWritable> values ,Context context)
throws IOException, InterruptedException{
int sum=0;
for(IntWritable value:values){
sum+=value.get();
}
temp.set("countCircle");
plus.set(sum);
context.write(temp, plus);
}

}

（4）Reducer：

public class piReduce extends Reducer<Text,IntWritable,Text,FloatWritable>{
Text pi=new Text();
FloatWritable result=new FloatWritable();
public void reduce(Text key,Iterable<IntWritable> values ,Context context)
throws IOException, InterruptedException{
int countCircle=0;
for(IntWritable value:values){
System.out.println(value);
countCircle+=value.get();
}
float re=(float) (4*(countCircle/10000.0));//计算pi
pi.set("pi");
result.set(re);
context.write(pi, result);
}

}