You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by li...@apache.org on 2017/09/17 11:04:11 UTC
kylin git commit: KYLIN-2857 code review
Repository: kylin
Updated Branches:
refs/heads/ly e39d4e6ed -> 09e8bbe90
KYLIN-2857 code review
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/09e8bbe9
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/09e8bbe9
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/09e8bbe9
Branch: refs/heads/ly
Commit: 09e8bbe90e62f100dfa8fb142c81071fec5c4ee1
Parents: e39d4e6
Author: Li Yang <li...@apache.org>
Authored: Sun Sep 17 19:03:51 2017 +0800
Committer: Li Yang <li...@apache.org>
Committed: Sun Sep 17 19:03:51 2017 +0800
----------------------------------------------------------------------
.../engine/mr/common/MapReduceExecutable.java | 26 ++++----------------
1 file changed, 5 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/09e8bbe9/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/MapReduceExecutable.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/MapReduceExecutable.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/MapReduceExecutable.java
index 4e6458f..c2d1703 100755
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/MapReduceExecutable.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/MapReduceExecutable.java
@@ -80,10 +80,8 @@ public class MapReduceExecutable extends AbstractExecutable {
return;
}
try {
- String params = getMapReduceParams();
- String[] args = params.trim().split("\\s+");
- Configuration conf = HadoopUtil.getCurrentConfiguration();
- overwriteJobConf(conf, executableContext.getConfig(), args);
+ Configuration conf = new Configuration(HadoopUtil.getCurrentConfiguration());
+ overwriteJobConf(conf, executableContext.getConfig(), getMapReduceParams().trim().split("\\s+"));
Job job = new Cluster(conf).getJob(JobID.forName(mrJobId));
if (job == null || job.getJobState() == JobStatus.State.FAILED) {
//remove previous mr job info
@@ -107,15 +105,12 @@ public class MapReduceExecutable extends AbstractExecutable {
@Override
protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException {
final String mapReduceJobClass = getMapReduceJobClass();
- String params = getMapReduceParams();
Preconditions.checkNotNull(mapReduceJobClass);
- Preconditions.checkNotNull(params);
try {
Job job;
ExecutableManager mgr = getManager();
- String[] args = params.trim().split("\\s+");
- Configuration conf = HadoopUtil.getCurrentConfiguration();
- String[] jobArgs = overwriteJobConf(conf, context.getConfig(), args);
+ Configuration conf = new Configuration(HadoopUtil.getCurrentConfiguration());
+ String[] jobArgs = overwriteJobConf(conf, context.getConfig(), getMapReduceParams().trim().split("\\s+"));
final Map<String, String> extra = mgr.getOutput(getId()).getExtra();
if (extra.containsKey(ExecutableConstants.MR_JOB_ID)) {
job = new Cluster(conf).getJob(JobID.forName(extra.get(ExecutableConstants.MR_JOB_ID)));
@@ -125,12 +120,9 @@ public class MapReduceExecutable extends AbstractExecutable {
final AbstractHadoopJob hadoopJob = constructor.newInstance();
hadoopJob.setConf(conf);
hadoopJob.setAsync(true); // so the ToolRunner.run() returns right away
- logger.info("parameters of the MapReduceExecutable: {}", params);
+ logger.info("parameters of the MapReduceExecutable: {}", getMapReduceParams());
try {
- //for async mr job, ToolRunner just return 0;
- // use this method instead of ToolRunner.run() because ToolRunner.run() is not thread-sale
- // Refer to: http://stackoverflow.com/questions/22462665/is-hadoops-toorunner-thread-safe
hadoopJob.run(jobArgs);
if (hadoopJob.isSkipped()) {
@@ -150,14 +142,6 @@ public class MapReduceExecutable extends AbstractExecutable {
final StringBuilder output = new StringBuilder();
final HadoopCmdOutput hadoopCmdOutput = new HadoopCmdOutput(job, output);
- // final String restStatusCheckUrl = getRestStatusCheckUrl(job, context.getConfig());
- // if (restStatusCheckUrl == null) {
- // logger.error("restStatusCheckUrl is null");
- // return new ExecuteResult(ExecuteResult.State.ERROR, "restStatusCheckUrl is null");
- // }
- // String mrJobId = hadoopCmdOutput.getMrJobId();
- // boolean useKerberosAuth = context.getConfig().isGetJobStatusWithKerberos();
- // HadoopStatusChecker statusChecker = new HadoopStatusChecker(restStatusCheckUrl, mrJobId, output, useKerberosAuth);
JobStepStatusEnum status = JobStepStatusEnum.NEW;
while (!isDiscarded() && !isPaused()) {