@Test public void testOverwriteJobConf() throws Exception { MapReduceExecutable executable = new MapReduceExecutable(); KylinConfig config = KylinConfig.getInstanceFromEnv(); Method method = MapReduceExecutable.class.getDeclaredMethod("overwriteJobConf", Configuration.class, KylinConfig.class, new String[] {}.getClass()); method.setAccessible(true); Configuration conf = new Configuration(); conf.set("mapreduce.job.is-mem-hungry", "true"); method.invoke(executable, conf, config, new String[] { "-cubename", "ci_inner_join_cube" }); Assert.assertEquals("mem-test1", conf.get("test1")); Assert.assertEquals("mem-test2", conf.get("test2")); conf.clear(); method.invoke(executable, conf, config, new String[] { "-cubename", "ci_inner_join_cube" }); Assert.assertEquals("test1", conf.get("test1")); Assert.assertEquals("test2", conf.get("test2")); }
private String[] overwriteJobConf(Configuration conf, KylinConfig config, String[] jobParams) throws ParseException { Options options = new Options(); options.addOption(OPTION_JOB_CONF); options.addOption(OPTION_CUBE_NAME); CustomParser parser = new CustomParser(); CommandLine commandLine = parser.parse(options, jobParams); String confFile = commandLine.getOptionValue(BatchConstants.ARG_CONF); String cubeName = commandLine.getOptionValue(BatchConstants.ARG_CUBE_NAME); List<String> remainingArgs = Lists.newArrayList(); if (StringUtils.isNotBlank(confFile)) { conf.addResource(new Path(confFile)); } KylinConfig configOverride; if (cubeName != null) { configOverride = CubeManager.getInstance(config).getCube(cubeName).getConfig(); } else { configOverride = config; } for (Map.Entry<String, String> entry : configOverride.getMRConfigOverride().entrySet()) { conf.set(entry.getKey(), entry.getValue()); } if (conf.get("mapreduce.job.is-mem-hungry") != null && Boolean.parseBoolean(conf.get("mapreduce.job.is-mem-hungry"))) { for (Map.Entry<String, String> entry : configOverride.getMemHungryConfigOverride().entrySet()) { conf.set(entry.getKey(), entry.getValue()); } } if (StringUtils.isNotBlank(cubeName)) { remainingArgs.add("-" + BatchConstants.ARG_CUBE_NAME); remainingArgs.add(cubeName); } remainingArgs.addAll(parser.getRemainingArgs()); return (String[]) remainingArgs.toArray(new String[remainingArgs.size()]); }
MapReduceExecutable extends AbstractExecutable { private String[] overwriteJobConf(Configuration conf, KylinConfig config, String[] jobParams) throws ParseException { Options options = new Options(); options.addOption(OPTION_JOB_CONF); options.addOption(OPTION_CUBE_NAME); CustomParser parser = new CustomParser(); CommandLine commandLine = parser.parse(options, jobParams); String confFile = commandLine.getOptionValue(BatchConstants.ARG_CONF); String cubeName = commandLine.getOptionValue(BatchConstants.ARG_CUBE_NAME); List<String> remainingArgs = Lists.newArrayList(); if (StringUtils.isNotBlank(confFile)) { conf.addResource(new Path(confFile)); } KylinConfig configOverride; if (cubeName != null) { configOverride = CubeManager.getInstance(config).getCube(cubeName).getConfig(); } else { configOverride = config; } for (Map.Entry<String, String> entry : configOverride.getMRConfigOverride().entrySet()) { conf.set(entry.getKey(), entry.getValue()); } if (conf.get("mapreduce.job.is-mem-hungry") != null && Boolean.parseBoolean(conf.get("mapreduce.job.is-mem-hungry"))) { for (Map.Entry<String, String> entry : configOverride.getMemHungryConfigOverride().entrySet()) { conf.set(entry.getKey(), entry.getValue()); } } if (StringUtils.isNotBlank(cubeName)) { remainingArgs.add("-" + BatchConstants.ARG_CUBE_NAME); remainingArgs.add(cubeName); } remainingArgs.addAll(parser.getRemainingArgs()); return (String[]) remainingArgs.toArray(new String[remainingArgs.size()]); } }
MapReduceExecutable extends AbstractExecutable { private String[] overwriteJobConf(Configuration conf, KylinConfig config, String[] jobParams) throws ParseException { Options options = new Options(); options.addOption(OPTION_JOB_CONF); options.addOption(OPTION_CUBE_NAME); CustomParser parser = new CustomParser(); CommandLine commandLine = parser.parse(options, jobParams); String confFile = commandLine.getOptionValue(BatchConstants.ARG_CONF); String cubeName = commandLine.getOptionValue(BatchConstants.ARG_CUBE_NAME); List<String> remainingArgs = Lists.newArrayList(); if (StringUtils.isNotBlank(confFile)) { conf.addResource(new Path(confFile)); } KylinConfig configOverride; if (cubeName != null) { configOverride = CubeManager.getInstance(config).getCube(cubeName).getConfig(); } else { configOverride = config; } for (Map.Entry<String, String> entry : configOverride.getMRConfigOverride().entrySet()) { conf.set(entry.getKey(), entry.getValue()); } if (conf.get("mapreduce.job.is-mem-hungry") != null && Boolean.parseBoolean(conf.get("mapreduce.job.is-mem-hungry"))) { for (Map.Entry<String, String> entry : configOverride.getMemHungryConfigOverride().entrySet()) { conf.set(entry.getKey(), entry.getValue()); } } if (StringUtils.isNotBlank(cubeName)) { remainingArgs.add("-" + BatchConstants.ARG_CUBE_NAME); remainingArgs.add(cubeName); } remainingArgs.addAll(parser.getRemainingArgs()); return (String[]) remainingArgs.toArray(new String[remainingArgs.size()]); } MapReduceExecutable(); }
MapReduceExecutable extends AbstractExecutable { private String[] overwriteJobConf(Configuration conf, KylinConfig config, String[] jobParams) throws ParseException { Options options = new Options(); options.addOption(OPTION_JOB_CONF); options.addOption(OPTION_CUBE_NAME); CustomParser parser = new CustomParser(); CommandLine commandLine = parser.parse(options, jobParams); String confFile = commandLine.getOptionValue(BatchConstants.ARG_CONF); String cubeName = commandLine.getOptionValue(BatchConstants.ARG_CUBE_NAME); List<String> remainingArgs = Lists.newArrayList(); if (StringUtils.isNotBlank(confFile)) { conf.addResource(new Path(confFile)); } KylinConfig configOverride; if (cubeName != null) { configOverride = CubeManager.getInstance(config).getCube(cubeName).getConfig(); } else { configOverride = config; } for (Map.Entry<String, String> entry : configOverride.getMRConfigOverride().entrySet()) { conf.set(entry.getKey(), entry.getValue()); } if (conf.get("mapreduce.job.is-mem-hungry") != null && Boolean.parseBoolean(conf.get("mapreduce.job.is-mem-hungry"))) { for (Map.Entry<String, String> entry : configOverride.getMemHungryConfigOverride().entrySet()) { conf.set(entry.getKey(), entry.getValue()); } } if (StringUtils.isNotBlank(cubeName)) { remainingArgs.add("-" + BatchConstants.ARG_CUBE_NAME); remainingArgs.add(cubeName); } remainingArgs.addAll(parser.getRemainingArgs()); return (String[]) remainingArgs.toArray(new String[remainingArgs.size()]); } MapReduceExecutable(); long getMapReduceWaitTime(); void setMapReduceWaitTime(long t); String getMapReduceJobClass(); void setMapReduceJobClass(Class<? extends AbstractHadoopJob> clazzName); String getMapReduceParams(); void setMapReduceParams(String param); void setCounterSaveAs(String value); void setIsNeedLock(Boolean isNeedLock); boolean getIsNeedLock(); void setIsNeedReleaseLock(Boolean isNeedReleaseLock); boolean getIsNeedReleaseLock(); void setLockPathName(String pathName); String getLockPathName(); void setJobFlowJobId(String jobId); String getJobFlowJobId(); }
MapReduceExecutable extends AbstractExecutable { private String[] overwriteJobConf(Configuration conf, KylinConfig config, String[] jobParams) throws ParseException { Options options = new Options(); options.addOption(OPTION_JOB_CONF); options.addOption(OPTION_CUBE_NAME); CustomParser parser = new CustomParser(); CommandLine commandLine = parser.parse(options, jobParams); String confFile = commandLine.getOptionValue(BatchConstants.ARG_CONF); String cubeName = commandLine.getOptionValue(BatchConstants.ARG_CUBE_NAME); List<String> remainingArgs = Lists.newArrayList(); if (StringUtils.isNotBlank(confFile)) { conf.addResource(new Path(confFile)); } KylinConfig configOverride; if (cubeName != null) { configOverride = CubeManager.getInstance(config).getCube(cubeName).getConfig(); } else { configOverride = config; } for (Map.Entry<String, String> entry : configOverride.getMRConfigOverride().entrySet()) { conf.set(entry.getKey(), entry.getValue()); } if (conf.get("mapreduce.job.is-mem-hungry") != null && Boolean.parseBoolean(conf.get("mapreduce.job.is-mem-hungry"))) { for (Map.Entry<String, String> entry : configOverride.getMemHungryConfigOverride().entrySet()) { conf.set(entry.getKey(), entry.getValue()); } } if (StringUtils.isNotBlank(cubeName)) { remainingArgs.add("-" + BatchConstants.ARG_CUBE_NAME); remainingArgs.add(cubeName); } remainingArgs.addAll(parser.getRemainingArgs()); return (String[]) remainingArgs.toArray(new String[remainingArgs.size()]); } MapReduceExecutable(); long getMapReduceWaitTime(); void setMapReduceWaitTime(long t); String getMapReduceJobClass(); void setMapReduceJobClass(Class<? extends AbstractHadoopJob> clazzName); String getMapReduceParams(); void setMapReduceParams(String param); void setCounterSaveAs(String value); void setIsNeedLock(Boolean isNeedLock); boolean getIsNeedLock(); void setIsNeedReleaseLock(Boolean isNeedReleaseLock); boolean getIsNeedReleaseLock(); void setLockPathName(String pathName); String getLockPathName(); void setJobFlowJobId(String jobId); String getJobFlowJobId(); static final String MAP_REDUCE_WAIT_TIME; }