hadoop - map() 函数的调用次数与 MR Job 发出的 map 任务数之间的关系

Question

我写了一个MR程序来估计PI(3.141592.........)如下，但是我遇到了一个问题:

框架发出的 map 任务数是 11，下面是输出(总共 35 行)。但我预计输出是 11 行。有什么我想念的吗？

圆圈 78534096围圈 78539304圈子78540871围圈 78537925圈子 78537161圈子 78544419围圈 78537045圈子 78534861圈子 78545779圆圈 78528890围圈 78540007围圈 78542686圈子 78534539圈子 78538255圈子 78543392圈子78543191围圈 78540938圈子 78534882围圈 78536155圈子 78545739围圈 78541807围圈 78540635圈子78547561圈子 78540521围圈 78541320圈子 78537605圈子78541379圆环 78540408圈子 78536238围圈 78539614圈子78539773圈子 78537169围圈 78541707围圈 78537141圆圈 78538045

//程序开始导入...

公共(public)类 PiEstimation {

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

            private final static Text  INCIRCLE             = new Text("INCIRCLE");
            private final static LongWritable TimesInAMap   = new LongWritable(100000000);
            private static Random random = new Random();

            public  class MyPoint {
                    private double  x = 0.0;
                    private double  y = 0.0;

                    MyPoint(double _x,double _y) {
                            this.x = _x;
                            this.y = _y;
                    }

                    public boolean inCircle() {
                            if ( ((x-0.5)*(x-0.5) + (y-0.5)*(y-0.5)) <= 0.25 )
                                    return true;
                            else
                                    return false;
                    }

                    public void setPoint(double _x,double _y) {
                            this.x = _x;
                            this.y = _y;
                    }
            }
            public void map(LongWritable key, Text value, OutputCollector<Text, LongWritable> output, Reporter reporter) throws IOException {
                            long i = 0;
                            long N = TimesInAMap.get();
                            MyPoint myPoint = new MyPoint(random.nextDouble(),random.nextDouble());
                            long sum = 0;
                            while (i < N ) {
                            if (myPoint.inCircle()) {                                           
                                sum++;
                            }
                            myPoint.setPoint(random.nextDouble(),random.nextDouble());
                            i++;
                            }
                            output.collect(INCIRCLE, new LongWritable(sum));
                            }
            }


    public static class Reduce extends MapReduceBase implements Reducer<Text, LongWritable, Text, LongWritable> {
  public void reduce(Text key, Iterator<LongWritable> values, OutputCollector<Text, LongWritable> output, Reporter reporter) throws IOException {
      long sum = 0;
      while (values.hasNext()) {
        //sum += values.next().get();
        output.collect(key, values.next());
      }
      //output.collect(key, new LongWritable(sum));
  }
  }
    public static void main(String[] args) throws Exception {
  JobConf conf = new JobConf(PiEstimation.class);
  conf.setJobName("PiEstimation");

  conf.setOutputKeyClass(Text.class);
  conf.setOutputValueClass(LongWritable.class);

  conf.setMapperClass(Map.class);
  conf.setCombinerClass(Reduce.class);
  conf.setReducerClass(Reduce.class);

  conf.setInputFormat(TextInputFormat.class);
  conf.setOutputFormat(TextOutputFormat.class);
  conf.setNumMapTasks(10);
  conf.setNumReduceTasks(1);
  FileInputFormat.setInputPaths(conf, new Path(args[0]));
  FileOutputFormat.setOutputPath(conf, new Path(args[1]));

  JobClient.runJob(conf);
}

Answer 1

启动的 map 任务的数量由许多因素决定 - 主要是输入格式、将输入文件分块的相关 block 大小以及输入文件本身是否“可拆分”

另外，调用 map 的次数取决于每个 map 拆分中的记录数(mapper 正在处理的数据)。

假设您有一个 100 行的文本文件用于输入 - 很可能这将由一个 Mapper 处理，但是 map 方法被调用了 100 次 - 输入文件中的每一行一次

如果您计算输入文件中的行数 - 即所有映射器调用映射的次数。很难准确确定每个 Mapper 将调用多少次 map。