@Test public void testMapReduce() throws Exception { Cluster cluster = newCluster(5); Configuration conf = HadoopConfigurationBuilder .buildMapReduceConfiguration(clusterSpec, cluster, defaults, Sets.newLinkedHashSet(Lists.newArrayList("/data0", "/data1"))); assertThat(conf.getString("p1"), is("mapred1")); assertThat(conf.getString("mapred.job.tracker"), matches(".+:8021")); assertThat(conf.getString("mapred.tasktracker.map.tasks.maximum"), is("4")); assertThat(conf.getString("mapred.tasktracker.reduce.tasks.maximum"), is("3")); assertThat(conf.getString("mapred.reduce.tasks"), is("15")); assertThat(conf.getString("mapred.local.dir"), is("/data0/hadoop/mapred/local,/data1/hadoop/mapred/local")); }
@VisibleForTesting static Configuration buildMapReduceConfiguration(ClusterSpec clusterSpec, Cluster cluster, Configuration defaults, Set<String> dataDirectories) throws ConfigurationException, IOException { Configuration config = build(clusterSpec, cluster, defaults, "hadoop-mapreduce"); setIfAbsent(config, "mapred.local.dir", appendToDataDirectories(dataDirectories, "/hadoop/mapred/local")); Set<Instance> taskTrackers = cluster .getInstancesMatching(role(HadoopTaskTrackerClusterActionHandler.ROLE)); if (!taskTrackers.isEmpty()) { Hardware hardware = Iterables.getFirst(taskTrackers, null) .getNodeMetadata().getHardware(); if (hardware != null) { int coresPerNode = 0; for (Processor processor : hardware.getProcessors()) { coresPerNode += processor.getCores(); } int mapTasksPerNode = (int) Math.ceil(coresPerNode * 1.0); int reduceTasksPerNode = (int) Math.ceil(coresPerNode * 0.75); setIfAbsent(config, "mapred.tasktracker.map.tasks.maximum", mapTasksPerNode + ""); setIfAbsent(config, "mapred.tasktracker.reduce.tasks.maximum", reduceTasksPerNode + ""); int clusterReduceSlots = taskTrackers.size() * reduceTasksPerNode; setIfAbsent(config, "mapred.reduce.tasks", clusterReduceSlots + ""); } } Set<Instance> jobtracker = cluster .getInstancesMatching(role(HadoopJobTrackerClusterActionHandler.ROLE)); if (!jobtracker.isEmpty()) { config.setProperty("mapred.job.tracker", String.format("%s:8021", Iterables.getOnlyElement(jobtracker).getPublicHostName())); } return config; }
HadoopConfigurationBuilder { @VisibleForTesting static Configuration buildMapReduceConfiguration(ClusterSpec clusterSpec, Cluster cluster, Configuration defaults, Set<String> dataDirectories) throws ConfigurationException, IOException { Configuration config = build(clusterSpec, cluster, defaults, "hadoop-mapreduce"); setIfAbsent(config, "mapred.local.dir", appendToDataDirectories(dataDirectories, "/hadoop/mapred/local")); Set<Instance> taskTrackers = cluster .getInstancesMatching(role(HadoopTaskTrackerClusterActionHandler.ROLE)); if (!taskTrackers.isEmpty()) { Hardware hardware = Iterables.getFirst(taskTrackers, null) .getNodeMetadata().getHardware(); if (hardware != null) { int coresPerNode = 0; for (Processor processor : hardware.getProcessors()) { coresPerNode += processor.getCores(); } int mapTasksPerNode = (int) Math.ceil(coresPerNode * 1.0); int reduceTasksPerNode = (int) Math.ceil(coresPerNode * 0.75); setIfAbsent(config, "mapred.tasktracker.map.tasks.maximum", mapTasksPerNode + ""); setIfAbsent(config, "mapred.tasktracker.reduce.tasks.maximum", reduceTasksPerNode + ""); int clusterReduceSlots = taskTrackers.size() * reduceTasksPerNode; setIfAbsent(config, "mapred.reduce.tasks", clusterReduceSlots + ""); } } Set<Instance> jobtracker = cluster .getInstancesMatching(role(HadoopJobTrackerClusterActionHandler.ROLE)); if (!jobtracker.isEmpty()) { config.setProperty("mapred.job.tracker", String.format("%s:8021", Iterables.getOnlyElement(jobtracker).getPublicHostName())); } return config; } }
HadoopConfigurationBuilder { @VisibleForTesting static Configuration buildMapReduceConfiguration(ClusterSpec clusterSpec, Cluster cluster, Configuration defaults, Set<String> dataDirectories) throws ConfigurationException, IOException { Configuration config = build(clusterSpec, cluster, defaults, "hadoop-mapreduce"); setIfAbsent(config, "mapred.local.dir", appendToDataDirectories(dataDirectories, "/hadoop/mapred/local")); Set<Instance> taskTrackers = cluster .getInstancesMatching(role(HadoopTaskTrackerClusterActionHandler.ROLE)); if (!taskTrackers.isEmpty()) { Hardware hardware = Iterables.getFirst(taskTrackers, null) .getNodeMetadata().getHardware(); if (hardware != null) { int coresPerNode = 0; for (Processor processor : hardware.getProcessors()) { coresPerNode += processor.getCores(); } int mapTasksPerNode = (int) Math.ceil(coresPerNode * 1.0); int reduceTasksPerNode = (int) Math.ceil(coresPerNode * 0.75); setIfAbsent(config, "mapred.tasktracker.map.tasks.maximum", mapTasksPerNode + ""); setIfAbsent(config, "mapred.tasktracker.reduce.tasks.maximum", reduceTasksPerNode + ""); int clusterReduceSlots = taskTrackers.size() * reduceTasksPerNode; setIfAbsent(config, "mapred.reduce.tasks", clusterReduceSlots + ""); } } Set<Instance> jobtracker = cluster .getInstancesMatching(role(HadoopJobTrackerClusterActionHandler.ROLE)); if (!jobtracker.isEmpty()) { config.setProperty("mapred.job.tracker", String.format("%s:8021", Iterables.getOnlyElement(jobtracker).getPublicHostName())); } return config; }  }
HadoopConfigurationBuilder { @VisibleForTesting static Configuration buildMapReduceConfiguration(ClusterSpec clusterSpec, Cluster cluster, Configuration defaults, Set<String> dataDirectories) throws ConfigurationException, IOException { Configuration config = build(clusterSpec, cluster, defaults, "hadoop-mapreduce"); setIfAbsent(config, "mapred.local.dir", appendToDataDirectories(dataDirectories, "/hadoop/mapred/local")); Set<Instance> taskTrackers = cluster .getInstancesMatching(role(HadoopTaskTrackerClusterActionHandler.ROLE)); if (!taskTrackers.isEmpty()) { Hardware hardware = Iterables.getFirst(taskTrackers, null) .getNodeMetadata().getHardware(); if (hardware != null) { int coresPerNode = 0; for (Processor processor : hardware.getProcessors()) { coresPerNode += processor.getCores(); } int mapTasksPerNode = (int) Math.ceil(coresPerNode * 1.0); int reduceTasksPerNode = (int) Math.ceil(coresPerNode * 0.75); setIfAbsent(config, "mapred.tasktracker.map.tasks.maximum", mapTasksPerNode + ""); setIfAbsent(config, "mapred.tasktracker.reduce.tasks.maximum", reduceTasksPerNode + ""); int clusterReduceSlots = taskTrackers.size() * reduceTasksPerNode; setIfAbsent(config, "mapred.reduce.tasks", clusterReduceSlots + ""); } } Set<Instance> jobtracker = cluster .getInstancesMatching(role(HadoopJobTrackerClusterActionHandler.ROLE)); if (!jobtracker.isEmpty()) { config.setProperty("mapred.job.tracker", String.format("%s:8021", Iterables.getOnlyElement(jobtracker).getPublicHostName())); } return config; }  static Statement buildCommon(String path, ClusterSpec clusterSpec,
      Cluster cluster); static Statement buildHdfs(String path, ClusterSpec clusterSpec,
      Cluster cluster, Set<String> dataDirectories); static Statement buildMapReduce(String path, ClusterSpec clusterSpec,
      Cluster cluster, Set<String> dataDirectories); static Statement buildHadoopEnv(String path, ClusterSpec clusterSpec,
      Cluster cluster); }
HadoopConfigurationBuilder { @VisibleForTesting static Configuration buildMapReduceConfiguration(ClusterSpec clusterSpec, Cluster cluster, Configuration defaults, Set<String> dataDirectories) throws ConfigurationException, IOException { Configuration config = build(clusterSpec, cluster, defaults, "hadoop-mapreduce"); setIfAbsent(config, "mapred.local.dir", appendToDataDirectories(dataDirectories, "/hadoop/mapred/local")); Set<Instance> taskTrackers = cluster .getInstancesMatching(role(HadoopTaskTrackerClusterActionHandler.ROLE)); if (!taskTrackers.isEmpty()) { Hardware hardware = Iterables.getFirst(taskTrackers, null) .getNodeMetadata().getHardware(); if (hardware != null) { int coresPerNode = 0; for (Processor processor : hardware.getProcessors()) { coresPerNode += processor.getCores(); } int mapTasksPerNode = (int) Math.ceil(coresPerNode * 1.0); int reduceTasksPerNode = (int) Math.ceil(coresPerNode * 0.75); setIfAbsent(config, "mapred.tasktracker.map.tasks.maximum", mapTasksPerNode + ""); setIfAbsent(config, "mapred.tasktracker.reduce.tasks.maximum", reduceTasksPerNode + ""); int clusterReduceSlots = taskTrackers.size() * reduceTasksPerNode; setIfAbsent(config, "mapred.reduce.tasks", clusterReduceSlots + ""); } } Set<Instance> jobtracker = cluster .getInstancesMatching(role(HadoopJobTrackerClusterActionHandler.ROLE)); if (!jobtracker.isEmpty()) { config.setProperty("mapred.job.tracker", String.format("%s:8021", Iterables.getOnlyElement(jobtracker).getPublicHostName())); } return config; }  static Statement buildCommon(String path, ClusterSpec clusterSpec,
      Cluster cluster); static Statement buildHdfs(String path, ClusterSpec clusterSpec,
      Cluster cluster, Set<String> dataDirectories); static Statement buildMapReduce(String path, ClusterSpec clusterSpec,
      Cluster cluster, Set<String> dataDirectories); static Statement buildHadoopEnv(String path, ClusterSpec clusterSpec,
      Cluster cluster);  }