You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by sh...@apache.org on 2015/09/21 03:18:14 UTC
incubator-kylin git commit: KYLIN-978 defer the deletion of the job
working on HBase cluster until segment be merged
Repository: incubator-kylin
Updated Branches:
refs/heads/1.x-staging faf401a0f -> 607aeb4fe
KYLIN-978 defer the deletion of the job working on HBase cluster until segment be merged
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/607aeb4f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/607aeb4f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/607aeb4f
Branch: refs/heads/1.x-staging
Commit: 607aeb4fed2eb6b992352f7a0324c52afdaaf2b5
Parents: faf401a
Author: shaofengshi <sh...@apache.org>
Authored: Mon Sep 21 09:17:50 2015 +0800
Committer: shaofengshi <sh...@apache.org>
Committed: Mon Sep 21 09:18:06 2015 +0800
----------------------------------------------------------------------
.../apache/kylin/job/cube/CubingJobBuilder.java | 29 ++++++++------------
.../kylin/job/cube/GarbageCollectionStep.java | 23 ++++------------
2 files changed, 17 insertions(+), 35 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/607aeb4f/job/src/main/java/org/apache/kylin/job/cube/CubingJobBuilder.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/job/cube/CubingJobBuilder.java b/job/src/main/java/org/apache/kylin/job/cube/CubingJobBuilder.java
index de75f7d..80c030f 100644
--- a/job/src/main/java/org/apache/kylin/job/cube/CubingJobBuilder.java
+++ b/job/src/main/java/org/apache/kylin/job/cube/CubingJobBuilder.java
@@ -63,11 +63,10 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
final CubingJob result = initialJob(seg, "BUILD");
final String jobId = result.getId();
final String cuboidRootPath = getJobWorkingDir(jobId) + "/" + seg.getCubeInstance().getName() + "/cuboid/";
- final List<String> toDeletePathsOnHadoopCluster = Lists.newArrayList();
- final List<String> toDeletePathsOnHBaseCluster = Lists.newArrayList();
+ final List<String> toDeletePaths = Lists.newArrayList();
// cubing
- Pair<AbstractExecutable, AbstractExecutable> twoSteps = addCubingSteps(seg, cuboidRootPath, result, toDeletePathsOnHadoopCluster);
+ Pair<AbstractExecutable, AbstractExecutable> twoSteps = addCubingSteps(seg, cuboidRootPath, result, toDeletePaths);
String intermediateHiveTableStepId = twoSteps.getFirst().getId();
String baseCuboidStepId = twoSteps.getSecond().getId();
@@ -79,8 +78,7 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
final CubeJoinedFlatTableDesc intermediateTableDesc = new CubeJoinedFlatTableDesc(seg.getCubeDesc(), seg);
final String hiveIntermediateTable = this.getIntermediateHiveTableName(intermediateTableDesc, jobId);
- toDeletePathsOnHBaseCluster.add(getJobWorkingDir(jobId));
- result.addTask(createGarbageCollectionStep(seg, null, hiveIntermediateTable, toDeletePathsOnHadoopCluster, toDeletePathsOnHBaseCluster));
+ result.addTask(createGarbageCollectionStep(seg, null, hiveIntermediateTable, toDeletePaths));
return result;
}
@@ -96,11 +94,10 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
List<String> mergingSegmentIds = Lists.newArrayList();
List<String> mergingCuboidPaths = Lists.newArrayList();
List<String> mergingHTables = Lists.newArrayList();
- final List<String> toDeletePathsOnHadoopCluster = Lists.newArrayList();
- final List<String> toDeletePathsOnHBaseCluster = Lists.newArrayList();
+ final List<String> toDeletePaths = Lists.newArrayList();
// cubing the incremental segment
- Pair<AbstractExecutable, AbstractExecutable> twoSteps = addCubingSteps(appendSegment, appendRootPath, result, toDeletePathsOnHadoopCluster);
+ Pair<AbstractExecutable, AbstractExecutable> twoSteps = addCubingSteps(appendSegment, appendRootPath, result, toDeletePaths);
final String intermediateHiveTableStepId = twoSteps.getFirst().getId();
final String baseCuboidStepId = twoSteps.getSecond().getId();
@@ -118,7 +115,7 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
} else {
mergingCuboidPaths.add(getPathToMerge(merging));
}
- toDeletePathsOnHadoopCluster.add(getJobWorkingDir(merging.getLastBuildJobID()));
+ toDeletePaths.add(getJobWorkingDir(merging.getLastBuildJobID()));
}
// merge cuboid
@@ -129,8 +126,7 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
// update cube info
result.addTask(createUpdateCubeInfoAfterMergeStep(mergeSegment, mergingSegmentIds, convertCuboidToHfileStep.getId(), jobId));
- toDeletePathsOnHBaseCluster.add(getJobWorkingDir(jobId));
- result.addTask(createGarbageCollectionStep(mergeSegment, mergingHTables, null, toDeletePathsOnHadoopCluster, toDeletePathsOnHBaseCluster));
+ result.addTask(createGarbageCollectionStep(mergeSegment, mergingHTables, null, toDeletePaths));
return result;
}
@@ -147,14 +143,13 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
List<String> mergingSegmentIds = Lists.newArrayList();
List<String> mergingCuboidPaths = Lists.newArrayList();
List<String> mergingHTables = Lists.newArrayList();
- final List<String> toDeletePathsOnHadoopCluster = Lists.newArrayList();
- final List<String> toDeletePathsOnHBaseCluster = Lists.newArrayList();
+ final List<String> toDeletePaths = Lists.newArrayList();
for (CubeSegment merging : mergingSegments) {
mergingSegmentIds.add(merging.getUuid());
mergingCuboidPaths.add(getPathToMerge(merging));
mergingHTables.add(merging.getStorageLocationIdentifier());
- toDeletePathsOnHadoopCluster.add(getJobWorkingDir(merging.getLastBuildJobID()));
+ toDeletePaths.add(getJobWorkingDir(merging.getLastBuildJobID()));
}
// merge cuboid
@@ -165,8 +160,7 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
// update cube info
result.addTask(createUpdateCubeInfoAfterMergeStep(seg, mergingSegmentIds, convertCuboidToHfileStep.getId(), jobId));
- toDeletePathsOnHBaseCluster.add(getJobWorkingDir(jobId));
- result.addTask(createGarbageCollectionStep(seg, mergingHTables, null, toDeletePathsOnHadoopCluster, toDeletePathsOnHBaseCluster));
+ result.addTask(createGarbageCollectionStep(seg, mergingHTables, null, toDeletePaths));
return result;
}
@@ -470,13 +464,12 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
return result;
}
- private GarbageCollectionStep createGarbageCollectionStep(CubeSegment seg, List<String> oldHtables, String hiveIntermediateTable, List<String> oldHdsfPaths, List<String> oldHdfsPathsOnHBaseCluster) {
+ private GarbageCollectionStep createGarbageCollectionStep(CubeSegment seg, List<String> oldHtables, String hiveIntermediateTable, List<String> oldHdsfPaths) {
GarbageCollectionStep result = new GarbageCollectionStep();
result.setName(ExecutableConstants.STEP_NAME_GARBAGE_COLLECTION);
result.setOldHTables(oldHtables);
result.setOldHiveTable(hiveIntermediateTable);
result.setOldHdfsPaths(oldHdsfPaths);
- result.setOldHdfsPathsOnHBaseCluster(oldHdfsPathsOnHBaseCluster);
return result;
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/607aeb4f/job/src/main/java/org/apache/kylin/job/cube/GarbageCollectionStep.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/job/cube/GarbageCollectionStep.java b/job/src/main/java/org/apache/kylin/job/cube/GarbageCollectionStep.java
index 641454c..f2f1fc0 100644
--- a/job/src/main/java/org/apache/kylin/job/cube/GarbageCollectionStep.java
+++ b/job/src/main/java/org/apache/kylin/job/cube/GarbageCollectionStep.java
@@ -55,8 +55,6 @@ public class GarbageCollectionStep extends AbstractExecutable {
private static final String OLD_HDFS_PATHS = "oldHdfsPaths";
- private static final String OLD_HDFS_PATHS_ON_HBASE_CLUSTER = "oldHdfsPathsOnHBaseCluster";
-
private static final Logger logger = LoggerFactory.getLogger(GarbageCollectionStep.class);
private StringBuffer output;
@@ -73,7 +71,6 @@ public class GarbageCollectionStep extends AbstractExecutable {
dropHBaseTable(context);
dropHiveTable(context);
dropHdfsPath(context);
- dropHdfsPathOnHBaseCluster(context);
} catch (IOException e) {
logger.error("job:" + getId() + " execute finished with exception", e);
output.append("\n").append(e.getLocalizedMessage());
@@ -160,12 +157,12 @@ public class GarbageCollectionStep extends AbstractExecutable {
List<String> oldHdfsPaths = this.getOldHdfsPaths();
FileSystem fileSystem = FileSystem.get(HadoopUtil.getCurrentConfiguration());
dropHdfsPathOnCluster(oldHdfsPaths, fileSystem);
- }
-
- private void dropHdfsPathOnHBaseCluster(ExecutableContext context) throws IOException {
- List<String> oldHdfsPaths = this.getOldHdfsPathsOnHBaseCluster();
- FileSystem fileSystem = FileSystem.get(HadoopUtil.getCurrentHBaseConfiguration());
- dropHdfsPathOnCluster(oldHdfsPaths, fileSystem);
+
+ if (StringUtils.isNotEmpty(context.getConfig().getHBaseClusterFs())) {
+ fileSystem = FileSystem.get(HadoopUtil.getCurrentHBaseConfiguration());
+ dropHdfsPathOnCluster(oldHdfsPaths, fileSystem);
+ }
+
}
public void setOldHTables(List<String> tables) {
@@ -184,14 +181,6 @@ public class GarbageCollectionStep extends AbstractExecutable {
return getArrayParam(OLD_HDFS_PATHS);
}
- public void setOldHdfsPathsOnHBaseCluster(List<String> paths) {
- setArrayParam(OLD_HDFS_PATHS_ON_HBASE_CLUSTER, paths);
- }
-
- private List<String> getOldHdfsPathsOnHBaseCluster() {
- return getArrayParam(OLD_HDFS_PATHS_ON_HBASE_CLUSTER);
- }
-
private void setArrayParam(String paramKey, List<String> paramValues) {
setParam(paramKey, StringUtils.join(paramValues, ","));
}