L3ProductionType extends HadoopProductionType { @Override public Production createProduction(ProductionRequest productionRequest) throws ProductionException { final String productionId = Production.createId(productionRequest.getProductionType()); String defaultProductionName = createProductionName("Level-3 ", productionRequest); final String productionName = productionRequest.getProductionName(defaultProductionName); List<DateRange> dateRanges = getDateRanges(productionRequest, null); if (dateRanges.size() == 0) { throw new ProductionException("No time ranges specified."); } ProcessorProductionRequest processorProductionRequest = new ProcessorProductionRequest(productionRequest); Geometry regionGeometry = productionRequest.getRegionGeometry(null); Integer numReducers = productionRequest.getInteger(JobConfigNames.CALVALUS_L3_REDUCERS, null); boolean singleReducer = numReducers != null && numReducers == 1; String outputFormat = productionRequest.getString("outputFormat", productionRequest.getString( JobConfigNames.CALVALUS_OUTPUT_FORMAT, null)); String outputDir = getOutputPath(productionRequest, productionId, "-L3-output"); boolean requiresFormatting = outputFormat != null && !outputFormat.equalsIgnoreCase("SEQ"); String l3ConfigXml = getL3ConfigXml(productionRequest); String[] l3OutputDirs = new String[dateRanges.size()]; Workflow workflow = new Workflow.Parallel(); workflow.setSustainable(false); for (int i = 0; i < dateRanges.size(); i++) { DateRange dateRange = dateRanges.get(i); String singleRangeOutputDir = getOutputPath(productionRequest, productionId, "-L3-" + (i + 1)); l3OutputDirs[i] = singleRangeOutputDir; Configuration jobConfig = createJobConfig(productionRequest); setDefaultProcessorParameters(processorProductionRequest, jobConfig); setRequestParameters(productionRequest, jobConfig); processorProductionRequest.configureProcessor(jobConfig); setInputLocationParameters(productionRequest, jobConfig); jobConfig.set(JobConfigNames.CALVALUS_INPUT_REGION_NAME, productionRequest.getRegionName()); jobConfig.set(JobConfigNames.CALVALUS_INPUT_DATE_RANGES, dateRange.toString()); jobConfig.set(JobConfigNames.CALVALUS_OUTPUT_DIR, singleRangeOutputDir); jobConfig.set(JobConfigNames.CALVALUS_L3_PARAMETERS, (jobConfig.get("calvalus.snap.bundle", "").contains("beam")) ? l3ConfigXml.replace("<planetaryGrid>org.esa.snap.binning.support.", "<planetaryGrid>org.esa.beam.binning.support.") : l3ConfigXml); jobConfig.set(JobConfigNames.CALVALUS_REGION_GEOMETRY, regionGeometry != null ? regionGeometry.toString() : ""); String date1Str = ProductionRequest.getDateFormat().format(dateRange.getStartDate()); String date2Str = ProductionRequest.getDateFormat().format(dateRange.getStopDate()); jobConfig.set(JobConfigNames.CALVALUS_MIN_DATE, date1Str); jobConfig.set(JobConfigNames.CALVALUS_MAX_DATE, date2Str); Integer superSampling = productionRequest.getInteger("superSampling", 1); if (superSampling > 1) { jobConfig.setBoolean("calvalus.system.snap.pixelGeoCoding.fractionAccuracy", true); jobConfig.setBoolean("calvalus.system.beam.pixelGeoCoding.fractionAccuracy", true); } if (singleReducer && requiresFormatting) { jobConfig.set(JobConfigNames.CALVALUS_OUTPUT_FORMAT, outputFormat); String outputCompression = productionRequest.getString("outputCompression", productionRequest.getString( JobConfigNames.CALVALUS_OUTPUT_COMPRESSION, "gz")); jobConfig.set(JobConfigNames.CALVALUS_OUTPUT_COMPRESSION, outputCompression); } WorkflowItem item = new L3WorkflowItem(getProcessingService(), productionRequest.getUserName(), productionName + " " + date1Str, jobConfig); workflow.add(item); } boolean hasQuicklookParameters = productionRequest.getString(JobConfigNames.CALVALUS_QUICKLOOK_PARAMETERS, null) != null; CalvalusLogger.getLogger().info("singleReducer=" + singleReducer + " requiresFormatting=" + requiresFormatting + " hasQuicklookParameters=" + hasQuicklookParameters); if (!singleReducer && requiresFormatting) { Configuration jobConfig = createJobConfig(productionRequest); setDefaultProcessorParameters(processorProductionRequest, jobConfig); setRequestParameters(productionRequest, jobConfig); jobConfig.unset(JobConfigNames.CALVALUS_INPUT_FORMAT); processorProductionRequest.configureProcessor(jobConfig); jobConfig.setStrings(JobConfigNames.CALVALUS_INPUT_DIR, l3OutputDirs); jobConfig.set(JobConfigNames.CALVALUS_OUTPUT_DIR, outputDir); jobConfig.set(JobConfigNames.CALVALUS_OUTPUT_FORMAT, outputFormat); String outputCompression = productionRequest.getString("outputCompression", productionRequest.getString( JobConfigNames.CALVALUS_OUTPUT_COMPRESSION, "gz")); jobConfig.set(JobConfigNames.CALVALUS_OUTPUT_COMPRESSION, outputCompression); WorkflowItem formatItem = new L3FormatWorkflowItem(getProcessingService(), productionRequest.getUserName(), productionName + " Format", jobConfig); workflow = new Workflow.Sequential(workflow, formatItem); if (hasQuicklookParameters) { CalvalusLogger.getLogger().info("creating ql step for reducers+formatting"); Configuration qlJobConfig = createJobConfig(productionRequest); setDefaultProcessorParameters(processorProductionRequest, qlJobConfig); setRequestParameters(productionRequest, qlJobConfig); qlJobConfig.set(JobConfigNames.CALVALUS_INPUT_PATH_PATTERNS, outputDir + "/(?!_.*|part-[rm]-\\d+).*"); qlJobConfig.set(JobConfigNames.CALVALUS_INPUT_FORMAT, outputFormat); qlJobConfig.set(JobConfigNames.CALVALUS_OUTPUT_DIR, outputDir); WorkflowItem qlItem = new QLWorkflowItem(getProcessingService(), productionRequest.getUserName(), productionName + " RGB", qlJobConfig); workflow.add(qlItem); } } else if (singleReducer && requiresFormatting && hasQuicklookParameters) { CalvalusLogger.getLogger().info("creating ql step for single reducer+formatting"); Configuration qlJobConfig = createJobConfig(productionRequest); setDefaultProcessorParameters(processorProductionRequest, qlJobConfig); setRequestParameters(productionRequest, qlJobConfig); String[] qlInputDirs = new String[l3OutputDirs.length]; for (int i = 0; i < qlInputDirs.length; i++) { qlInputDirs[i] = l3OutputDirs[i] + "/(?!_.*|part-[rm]-\\d+).*"; } qlJobConfig.setStrings(JobConfigNames.CALVALUS_INPUT_PATH_PATTERNS, qlInputDirs); qlJobConfig.set(JobConfigNames.CALVALUS_INPUT_FORMAT, outputFormat); qlJobConfig.set(JobConfigNames.CALVALUS_OUTPUT_DIR, outputDir); WorkflowItem qlItem = new QLWorkflowItem(getProcessingService(), productionRequest.getUserName(), productionName + " RGB", qlJobConfig); workflow = new Workflow.Sequential(workflow, qlItem); } String stagingDir = productionRequest.getStagingDirectory(productionId); boolean autoStaging = productionRequest.isAutoStaging(); return new Production(productionId, productionName, outputDir, stagingDir, autoStaging, productionRequest, workflow); } L3ProductionType(FileSystemService fileSystemService, HadoopProcessingService processingService,
                     StagingService stagingService); @Override Production createProduction(ProductionRequest productionRequest); static String getL3ConfigXml(ProductionRequest productionRequest);  }
@Test public void testCreateProduction() throws ProductionException, IOException { ProductionRequest productionRequest = createValidL3ProductionRequest(); Production production = productionType.createProduction(productionRequest); assertNotNull(production); assertEquals("Level-3 BandMaths 2010-06-15 to 2010-08-15", production.getName()); assertEquals(true, production.getStagingPath().startsWith("ewa/")); assertEquals(true, production.getId().contains("_" + "L3" + "_")); WorkflowItem workflow = production.getWorkflow(); assertNotNull(workflow); assertSame(Workflow.Sequential.class, workflow.getClass()); WorkflowItem[] sequenceWfs = workflow.getItems(); assertNotNull(sequenceWfs); assertEquals(2, sequenceWfs.length); WorkflowItem wfL3Parallel = sequenceWfs[0]; assertSame(Workflow.Parallel.class, wfL3Parallel.getClass()); WorkflowItem[] parallelWfs = wfL3Parallel.getItems(); assertNotNull(parallelWfs); assertEquals(3, parallelWfs.length); testL3Item(parallelWfs[0], "2010-06-15", "2010-06-19"); testL3Item(parallelWfs[1], "2010-07-05", "2010-07-09"); testL3Item(parallelWfs[2], "2010-07-25", "2010-07-29"); WorkflowItem wfFormat = sequenceWfs[1]; assertSame(L3FormatWorkflowItem.class, wfFormat.getClass()); L3FormatWorkflowItem l3formatWorkflowItem = (L3FormatWorkflowItem) wfFormat; assertEquals(true, l3formatWorkflowItem.getInputDir().startsWith("hdfs: assertEquals(true, l3formatWorkflowItem.getOutputDir().startsWith("hdfs: }