请给出hadoop自带计算圆周率的代码及注释

下面是 Hadoop 自带的计算圆周率的示例代码及注释：

package org.apache.hadoop.examples;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;

public class PiEstimator {

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    JobConf job = new JobConf(conf, PiEstimator.class);
    job.setJobName("pi");

    job.setOutputKeyClass(NullWritable.class);
    job.setOutputValueClass(NullWritable.class);

    job.setMapperClass(Map.class);
    job.setReducerClass(Reduce.class);

    job.setInputFormat(TextInputFormat.class);
    job.setOutputFormat(TextOutputFormat.class);

    FileInputFormat.setInputPaths(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));

    JobClient.runJob(job);
  }

  public static class Map extends MapReduceBase
      implements Mapper<Object, Text, NullWritable, NullWritable> {

    private long numRecords = 0;

    public void map(Object key, Text value,
        OutputCollector<NullWritable, NullWritable> output, Reporter reporter)
        throws IOException {

      String line = value.toString();
      int numInside = 0;
      int numOutside = 0;

      // 对于输入数据中的每一行，将其解析为两个 double 值，代表一个点的 x 和 y 坐标
      // 然后计算该点是否在圆内，并将计数器加 1
      // 最后输出两个计数器的值，供 Reduce 函数聚合
      for (String token : line.split("\\p{Space}+")) {
        double x = Double.parseDouble(token);
        double y = Double.parseDouble(token);
        if (x * x + y * y <= 1) {
          numInside++;
        } else {
          numOutside++;
        }
        numRecords++;
      }

      // 输出计数器的值，供 Reduce 函数聚合
      output.collect(NullWritable.get(), NullWritable.get());
    }

    @Override
    public void close() throws IOException {
      // 输出计数器的值，供 Reduce 函数聚合
      FileSystem fileSys = FileSystem.get(getConf());
      fileSys.create(new Path("mypath/num_records"), true).writeLong(numRecords);
    }
  }

  public static class Reduce extends MapReduceBase
      implements Reducer<NullWritable, NullWritable, NullWritable, DoubleWritable> {

    public void reduce(NullWritable key, Iterator<NullWritable> values,
        OutputCollector<NullWritable, DoubleWritable> output, Reporter reporter)
        throws IOException {
      Configuration conf = new Configuration();
      FileSystem fileSys = FileSystem.get(conf);
      long numRecords = fileSys.open(new Path("mypath/num_records")).readLong();

      // 计算圆周率的估计值
      double pi = 4.0 * numRecords / PiEstimator.NUM_MAP_TASKS / PiEstimator.NUM_SAMPLES_PER_MAP_TASK;

      // 输出圆周率的估计值
      DecimalFormat df = new DecimalFormat("#.##########");
      output.collect(NullWritable.get(), new DoubleWritable(Double.parseDouble(df.format(pi))));
    }
  }
  
  private static final int NUM_MAP_TASKS = 20;
  private static final int NUM_SAMPLES_PER_MAP_TASK = 100000;
}

这个示例代码包括一个 Mapper 和一个 Reducer。Mapper 阶段会对输入数据中的每个点进行判断，看该点是否在圆内，然后输出计数器的值，供 Reducer 函数聚合。Reducer 阶段会读取 Mapper 输出的计数器值，计算圆周率的估计值，并输出。该示例代码使用了 Hadoop 的 MapReduce 框架和 HDFS 分布式文件系统，可以通过输入参数指定输入和输出路径。