利用hadoop集群实现圆周率计算串行和并行的java完整代码

以下是利用Hadoop集群实现圆周率计算的串行和并行的Java完整代码。其中，串行的代码基于蒙特卡洛方法，而并行的代码使用了MapReduce框架。

## 串行代码

import java.util.Random;

public class PiSerial {
    public static void main(String[] args) {
        int nThrows = 100000000;
        Random rnd = new Random();
        int nSuccess = 0;
        for (int i = 0; i < nThrows; i++) {
            double x = rnd.nextDouble();
            double y = rnd.nextDouble();
            if (x * x + y * y <= 1) {
                nSuccess++;
            }
        }
        double pi = 4.0 * nSuccess / nThrows;
        System.out.println("Pi = " + pi);
    }
}

## 并行代码

### Mapper

import java.io.IOException;
import java.util.Random;

import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Mapper;

public class PiMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private int nThrows;

    protected void setup(Context context) throws IOException, InterruptedException {
        nThrows = context.getConfiguration().getInt("nThrows", 1000000);
    }

    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        Random rnd = new Random();
        int nSuccess = 0;
        for (int i = 0; i < nThrows; i++) {
            double x = rnd.nextDouble();
            double y = rnd.nextDouble();
            if (x * x + y * y <= 1) {
                nSuccess++;
            }
        }
        context.write(new Text("pi"), new IntWritable(nSuccess));
    }
}

### Reducer

import java.io.IOException;

import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Reducer;

public class PiReducer extends Reducer<Text, IntWritable, Text, DoubleWritable> {
    private int nThrows;

    protected void setup(Context context) throws IOException, InterruptedException {
        nThrows = context.getConfiguration().getInt("nThrows", 1000000);
    }

    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int nSuccess = 0;
        for (IntWritable value : values) {
            nSuccess += value.get();
        }
        double pi = 4.0 * nSuccess / (nThrows * context.getNumReduceTasks());
        context.write(key, new DoubleWritable(pi));
    }
}

### Driver

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class PiParallel {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        int nThrows = 100000000;
        conf.setInt("nThrows", nThrows);
        Job job = Job.getInstance(conf, "Pi Calculation Parallel");
        job.setJarByClass(PiParallel.class);
        job.setMapperClass(PiMapper.class);
        job.setReducerClass(PiReducer.class);
        job.setNumReduceTasks(4);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        TextInputFormat.addInputPath(job, new Path(args[0]));
        TextOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

请注意，在并行代码中，我们没有使用输入文件，而是使用了一个虚拟的输入，因为我们不需要输入任何数据，只需要在Mapper中生成随机数并计算圆周率即可。