MapReduceExecutor implements ExecutorPlugin { void run(final MapReduceExecutorTask task, int mapTaskCount, final int reduceTaskCount, final ProcessState state) { final ModelManager modelManager = task.getModelManager(); final Configuration conf = new Configuration(); for (String path : task.getConfigFiles()) { File file = new File(path); if (!file.isFile()) { throw new ConfigException(String.format("Config file '%s' does not exist", file)); } try { conf.addResource(file.toURI().toURL()); } catch (MalformedURLException ex) { throw new RuntimeException(ex); } } String uniqueTransactionName = getTransactionUniqueName(Exec.session()); final Path stateDir = new Path(new Path(task.getStatePath()), uniqueTransactionName); conf.set(MRJobConfig.MAPREDUCE_JOB_CLASSLOADER, "true"); conf.set(MRJobConfig.MAPREDUCE_JOB_CLASSLOADER_SYSTEM_CLASSES, "java.,org.apache.hadoop."); for (Map.Entry<String, String> pair : task.getConfig().entrySet()) { conf.set(pair.getKey(), pair.getValue()); } EmbulkMapReduce.setSystemConfig(conf, modelManager, systemConfig); EmbulkMapReduce.setExecutorTask(conf, modelManager, task); EmbulkMapReduce.setMapTaskCount(conf, mapTaskCount); EmbulkMapReduce.setRetryTasks(conf, task.getRetryTasks()); EmbulkMapReduce.setStateDirectoryPath(conf, stateDir); List<Path> jars = collectJars(task.getLibjars(), task.getExcludeJars()); conf.set("tmpjars", StringUtils.join(",", jars)); String remoteUser = conf.get(MRJobConfig.USER_NAME); if (remoteUser != null) { UserGroupInformation.createRemoteUser(remoteUser).doAs( new PrivilegedAction<Void>() { @Override public Void run() { runJob(task, modelManager, reduceTaskCount, state, stateDir, conf); return null; } } ); } else { runJob(task, modelManager, reduceTaskCount, state, stateDir, conf); } } @Inject  MapReduceExecutor(@ForSystemConfig ConfigSource systemConfig,
            ScriptingContainer jruby); @Override void transaction(ConfigSource config, Schema outputSchema, final int inputTaskCount,
            ExecutorPlugin.Control control);  }
@Test public void testEmbulkMapper() throws Exception { new File("tmp/embulk_mapred_output.000.00.csv").delete(); new File("tmp/embulk_mapred_output.001.00.csv").delete(); ConfigSource config = loadConfigSource(configLoader, "config/embulk_mapred_config.yml") .merge(configLoader.newConfigSource().set("exec", ImmutableMap.of("type", getPluginName()))); embulk.run(config); assertFileContent( Lists.newArrayList( RESOURCES_PATH + "fixtures/csv/sample1.csv", RESOURCES_PATH + "fixtures/csv/sample1.csv"), Lists.newArrayList( "tmp/embulk_mapred_output.000.00.csv", "tmp/embulk_mapred_output.001.00.csv")); }
@Test public void testEmbulkPartitioningMapperReducer() throws Exception { new File("tmp/embulk_mapred_partitioning_output.000.00.csv").delete(); new File("tmp/embulk_mapred_partitioning_output.001.00.csv").delete(); ConfigSource config = loadConfigSource(configLoader, "config/embulk_mapred_partitioning_config.yml") .merge(configLoader.newConfigSource().set("exec", ImmutableMap.of("type", getPluginName()))); embulk.run(config); assertFileContent( Lists.newArrayList( RESOURCES_PATH + "fixtures/csv/sample1.csv", RESOURCES_PATH + "fixtures/csv/sample1.csv"), Lists.newArrayList( "tmp/embulk_mapred_partitioning_output.000.00.csv", "tmp/embulk_mapred_partitioning_output.001.00.csv")); }
@Test public void testInvalidConfigFiles() throws Exception { try { ConfigSource config = loadConfigSource(configLoader, "config/embulk_mapred_invalid_config_files_config.yml") .merge(configLoader.newConfigSource().set("exec", ImmutableMap.of("type", getPluginName()))); embulk.run(config); fail(); } catch (Throwable t) { assertTrue(t instanceof PartialExecutionException); assertTrue(t.getCause() instanceof ConfigException); } }
@Test public void testInvalidPartitioning() throws Exception { try { ConfigSource config = loadConfigSource(configLoader, "config/embulk_mapred_invalid_partitioning_config.yml") .merge(configLoader.newConfigSource().set("exec", ImmutableMap.of("type", getPluginName()))); embulk.run(config); fail(); } catch (Throwable t) { assertTrue(t instanceof PartialExecutionException); assertTrue(t.getCause() instanceof ConfigException); } }
@Test public void testInvalidReducers() throws Exception { try { ConfigSource config = loadConfigSource(configLoader, "config/embulk_mapred_invalid_reducers_config.yml") .merge(configLoader.newConfigSource().set("exec", ImmutableMap.of("type", getPluginName()))); embulk.run(config); fail(); } catch (Throwable t) { assertTrue(t instanceof PartialExecutionException); assertTrue(t.getCause() instanceof ConfigException); } }
@Test public void testInvalidLibjars() throws Exception { try { ConfigSource config = loadConfigSource(configLoader, "config/embulk_mapred_invalid_libjars_config.yml") .merge(configLoader.newConfigSource().set("exec", ImmutableMap.of("type", getPluginName()))); embulk.run(config); fail(); } catch (Throwable t) { assertTrue(t instanceof PartialExecutionException); assertTrue(t.getCause() instanceof RuntimeException); assertTrue(t.getCause().getCause() instanceof FileNotFoundException); } }
@Test public void testStopOnInvalidRecord() throws Exception { try { ConfigSource config = loadConfigSource(configLoader, "config/embulk_mapred_stop_on_invalid_record_config.yml") .merge(configLoader.newConfigSource().set("exec", ImmutableMap.of("type", getPluginName()))); embulk.run(config); fail(); } catch (Throwable t) { t.printStackTrace(); assertTrue(t instanceof PartialExecutionException); assertTrue(UserDataExceptions.isUserDataException(t.getCause())); } }