java - 是否可以在一个节点上运行多个映射器

Question

我有 KMeans 代码，我的任务是计算加速比，我通过在我的 uni 集群中不同数量的节点上运行它来完成它。但是是否可以更改映射器和/或缩减器的数量，以便我可以在单个节点上运行时检查加速比的变化。

在谷歌搜索时，我发现通过使用 conf.setNumReduceTasks(2); 我可以更改 reducer 的数量。但我没有看到我的输出有任何变化。 (我的输出是以毫秒为单位的时间)。

我使用的代码来自github:https://github.com/himank/K-Means/blob/master/src/KMeans.java虽然我根据自己的要求做了一些改动，但是主要功能是一样的。

下面是 main 函数的样子:

    public static void main(String[] args) throws Exception {
    long startTime = System.currentTimeMillis();
    IN = args[0];
    OUT = args[1];
    String input = IN;
    String output = OUT + System.nanoTime();
    String again_input = output;
    int iteration = 0;
    boolean isdone = false;
    while (isdone == false) {
        JobConf conf = new JobConf(KMeans.class);
        if (iteration == 0) {
            Path hdfsPath = new Path(input + CENTROID_FILE_NAME);
            DistributedCache.addCacheFile(hdfsPath.toUri(), conf);
        } else {
            Path hdfsPath = new Path(again_input + OUTPUT_FILE_NAME);
            DistributedCache.addCacheFile(hdfsPath.toUri(), conf);
        }
        conf.setJobName(JOB_NAME);
        //conf.setNumReduceTasks(2);
        conf.setMapOutputKeyClass(DoubleWritable.class);
        conf.setMapOutputValueClass(DoubleWritable.class);
        conf.setOutputKeyClass(DoubleWritable.class);
        conf.setOutputValueClass(Text.class);
        conf.setMapperClass(Map.class);
        conf.setNumMapTasks(4);
        conf.setReducerClass(Reduce.class);
        conf.setInputFormat(TextInputFormat.class);
        conf.setOutputFormat(TextOutputFormat.class);
        FileInputFormat.setInputPaths(conf, new Path(input + DATA_FILE_NAME));
        FileOutputFormat.setOutputPath(conf, new Path(output));
        JobClient.runJob(conf);
        Path ofile = new Path(output + OUTPUT_FILE_NAME);   

        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://127.0.0.1:9000"), configuration);
        Path filePath = new Path(output + OUTPUT_FILE_NAME);
        BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(filePath)));
        List<Double> centers_next = new ArrayList<Double>();
        String line = br.readLine();
        while (line != null) {
            String[] sp = line.split("\t| ");
            double c = Double.parseDouble(sp[0]);
            centers_next.add(c);
            line = br.readLine();
        }
        br.close();
        String prev;
        if (iteration == 0) {
            prev = input + CENTROID_FILE_NAME;
        } else {
            prev = again_input + OUTPUT_FILE_NAME;
        }
        Path prevfile = new Path(prev);
        FileSystem fs1 = FileSystem.get(new URI("hdfs://127.0.0.1:9000"), configuration);
        BufferedReader br1 = new BufferedReader(new InputStreamReader(fs1.open(prevfile)));
        List<Double> centers_prev = new ArrayList<Double>();
        String l = br1.readLine();
        while (l != null) {
            String[] sp1 = l.split(SPLITTER);
            double d = Double.parseDouble(sp1[0]);
            centers_prev.add(d);
            l = br1.readLine();
        }
        br1.close();
        Collections.sort(centers_next);
        Collections.sort(centers_prev);
        Iterator<Double> it = centers_prev.iterator();
        for (double d : centers_next) {
            double temp = it.next();
            if (Math.abs(temp - d) <= 0.1) {
                isdone = true;
            } else {
                isdone = false;
                break;
            }
        }
        ++iteration;
        again_input = output;
        output = OUT + System.nanoTime();
    }
    long endTime   = System.currentTimeMillis();
    long totalTime = endTime - startTime;
    System.out.println(totalTime);
}

附言。我是 Hadoop 和 MapReduce 的新手。

Answer 1

给定作业的 map 数量通常由输入文件中的输入拆分数量驱动，而不是由 setNumMapTasks() 或 mapred.map.tasks 参数驱动。为每个输入拆分生成一个 Map 任务。 mapred.map.tasks 参数只是对 map 数量的 InputFormat 的提示。可以使用setNumMapTasks()手动增加map任务的数量，它可以用来增加map任务的数量，但不会设置低于Hadoop通过拆分输入数据确定的数量。

http://wiki.apache.org/hadoop/HowManyMapsAndReduces