MAWorkflowItem extends HadoopWorkflowItem { @Override protected void configureJob(Job job) throws IOException { Configuration jobConfig = job.getConfiguration(); jobConfig.setIfUnset("calvalus.system.snap.dataio.reader.tileWidth", "32"); jobConfig.setIfUnset("calvalus.system.snap.dataio.reader.tileHeight", "32"); jobConfig.setIfUnset("calvalus.system.snap.pixelGeoCoding.useTiling", "true"); jobConfig.setIfUnset("calvalus.system.snap.envisat.usePixelGeoCoding", "true"); jobConfig.setIfUnset("calvalus.system.snap.pixelGeoCoding.fractionAccuracy", "true"); jobConfig.setIfUnset("calvalus.system.beam.reader.tileWidth", "32"); jobConfig.setIfUnset("calvalus.system.beam.reader.tileHeight", "32"); jobConfig.setIfUnset("calvalus.system.beam.pixelGeoCoding.useTiling", "true"); jobConfig.setIfUnset("calvalus.system.beam.envisat.usePixelGeoCoding", "true"); jobConfig.setIfUnset("calvalus.system.beam.pixelGeoCoding.fractionAccuracy", "true"); jobConfig.setInt("mapred.job.reuse.jvm.num.tasks", 1); job.setInputFormatClass(getInputFormatClass(jobConfig)); job.setMapperClass(MAMapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(RecordWritable.class); job.setReducerClass(MAReducer.class); job.setNumReduceTasks(1); job.setOutputKeyClass(Text.class); job.setOutputValueClass(RecordWritable.class); job.setOutputFormatClass(SequenceFileOutputFormat.class); JobUtils.clearAndSetOutputDir(getOutputDir(), job, this); } MAWorkflowItem(HadoopProcessingService processingService, String username, String jobName, Configuration jobConfig); @Override String getOutputDir();  }
@Test public void testCreateJob() throws Exception { HadoopProcessingService processingService = new HadoopProcessingService(new JobClientsMap(new JobConf())); MAConfig maConfig = new MAConfig(); maConfig.setRecordSourceSpiClassName(TestRecordSourceSpi.class.getName()); String jobName = "MaTestJob"; Configuration jobConfig = processingService.createJobConfig("testuser"); jobConfig.set(JobConfigNames.CALVALUS_INPUT_PATH_PATTERNS, "file1,file2"); jobConfig.set(JobConfigNames.CALVALUS_INPUT_FORMAT, "HADOOP-STREAMING"); jobConfig.set(JobConfigNames.CALVALUS_OUTPUT_DIR, "out_004"); jobConfig.set(JobConfigNames.CALVALUS_MA_PARAMETERS, maConfig.toXml()); MAWorkflowItem maWorkflowItem = new MAWorkflowItem(processingService, "testuser", jobName, jobConfig); Job job = Job.getInstance(jobConfig); maWorkflowItem.configureJob(job); assertSame(MAMapper.class, job.getMapperClass()); assertSame(Text.class, job.getMapOutputKeyClass()); assertSame(RecordWritable.class, job.getMapOutputValueClass()); assertSame(SequenceFileOutputFormat.class, job.getOutputFormatClass()); assertEquals("out_004", FileOutputFormat.getOutputPath(job).getName()); }