MergeCuboidJob extends CuboidJob { @Override public int run(String[] args) throws Exception { Options options = new Options(); try { options.addOption(OPTION_JOB_NAME); options.addOption(OPTION_CUBE_NAME); options.addOption(OPTION_SEGMENT_ID); options.addOption(OPTION_INPUT_PATH); options.addOption(OPTION_OUTPUT_PATH); parseOptions(options, args); String input = getOptionValue(OPTION_INPUT_PATH); String output = getOptionValue(OPTION_OUTPUT_PATH); String cubeName = getOptionValue(OPTION_CUBE_NAME).toUpperCase(Locale.ROOT); String segmentID = getOptionValue(OPTION_SEGMENT_ID); CubeManager cubeMgr = CubeManager.getInstance(KylinConfig.getInstanceFromEnv()); CubeInstance cube = cubeMgr.getCube(cubeName); CubeSegment cubeSeg = cube.getSegmentById(segmentID); String jobName = getOptionValue(OPTION_JOB_NAME); logger.info("Starting: " + jobName); job = Job.getInstance(getConf(), jobName); setJobClasspath(job, cube.getConfig()); Segments<CubeSegment> allSegs = cube.getMergingSegments(cubeSeg); allSegs.add(cubeSeg); attachSegmentsMetadataWithDict(allSegs, job.getConfiguration()); job.setMapperClass(MergeCuboidMapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); job.setReducerClass(CuboidReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); IMROutput2.IMRMergeOutputFormat outputFormat = MRUtil.getBatchMergeOutputSide2(cubeSeg).getOutputFormat(); outputFormat.configureJobInput(job, input); addInputDirs(input, job); outputFormat.configureJobOutput(job, output, cubeSeg); job.getConfiguration().set(BatchConstants.CFG_CUBE_NAME, cubeName); job.getConfiguration().set(BatchConstants.CFG_CUBE_SEGMENT_ID, segmentID); return waitForCompletion(job); } finally { if (job != null) cleanupTempConfFile(job.getConfiguration()); } }  @Override int run(String[] args);  }
@Test public void test() throws Exception { String output = "target/test-output/merged_cuboid"; String cubeName = "test_kylin_cube_with_slr_ready"; String jobname = "merge_cuboid"; File baseFolder = File.createTempFile("kylin-f24668f6-dcff-4cb6-a89b-77f1119df8fa-", "base"); FileUtils.forceDelete(baseFolder); baseFolder.mkdir(); FileUtils.copyDirectory(new File("src/test/resources/data/base_cuboid"), baseFolder); FileUtils.forceDeleteOnExit(baseFolder); File eightFoler = File.createTempFile("kylin-f24668f6-dcff-4cb6-a89b-77f1119df8fa-", "8d"); FileUtils.forceDelete(eightFoler); eightFoler.mkdir(); FileUtils.copyDirectory(new File("src/test/resources/data/base_cuboid"), eightFoler); FileUtils.forceDeleteOnExit(eightFoler); FileUtil.fullyDelete(new File(output)); String[] args = { "-input", baseFolder.getAbsolutePath() + "," + eightFoler.getAbsolutePath(), "-cubename", cubeName, "-segmentname", "20130331080000_20131212080000", "-output", output, "-jobname", jobname }; assertEquals("Job failed", 0, ToolRunner.run(conf, new MergeCuboidJob(), args)); }