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/11 05:07:57 UTC
[1/2] incubator-kylin git commit: KYLIN-968
CubeSegment.lastBuildJobID is null in new instance but used for rowkey_stats
path
Repository: incubator-kylin
Updated Branches:
refs/heads/KYLIN-968-978 [created] bb1b1a725
KYLIN-968 CubeSegment.lastBuildJobID is null in new instance but used for rowkey_stats path
Signed-off-by: shaofengshi <sh...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/9df88498
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/9df88498
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/9df88498
Branch: refs/heads/KYLIN-968-978
Commit: 9df88498dd7fb3a732dcf9832098e43031c4cde7
Parents: 44309fe
Author: sunyerui <su...@gmail.com>
Authored: Thu Sep 10 11:23:16 2015 +0800
Committer: shaofengshi <sh...@apache.org>
Committed: Fri Sep 11 11:06:59 2015 +0800
----------------------------------------------------------------------
.../apache/kylin/job/cube/CubingJobBuilder.java | 16 +--
.../kylin/job/cube/CubingJobBuilderTest.java | 138 +++++++++++++++++++
2 files changed, 146 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/9df88498/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 5c3c277..ff79286 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
@@ -206,9 +206,9 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
final String jobId = result.getId();
final String cuboidPath = cuboidRootPath + "*";
- result.addTask(createRangeRowkeyDistributionStep(seg, cuboidPath));
+ result.addTask(createRangeRowkeyDistributionStep(seg, cuboidPath, jobId));
// create htable step
- result.addTask(createCreateHTableStep(seg));
+ result.addTask(createCreateHTableStep(seg, jobId));
// generate hfiles step
final MapReduceExecutable convertCuboidToHfileStep = createConvertCuboidToHfileStep(seg, cuboidPath, jobId);
result.addTask(convertCuboidToHfileStep);
@@ -265,8 +265,8 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
return getJobWorkingDir(seg.getLastBuildJobID()) + "/" + seg.getCubeInstance().getName() + "/cuboid/*";
}
- private String getRowkeyDistributionOutputPath(CubeSegment seg) {
- return getJobWorkingDir(seg.getLastBuildJobID()) + "/" + seg.getCubeInstance().getName() + "/rowkey_stats";
+ private String getRowkeyDistributionOutputPath(CubeSegment seg, String jobId) {
+ return getJobWorkingDir(jobId) + "/" + seg.getCubeInstance().getName() + "/rowkey_stats";
}
private String getFactDistinctColumnsPath(CubeSegment seg, String jobUuid) {
@@ -347,14 +347,14 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
return ndCuboidStep;
}
- private MapReduceExecutable createRangeRowkeyDistributionStep(CubeSegment seg, String inputPath) {
+ private MapReduceExecutable createRangeRowkeyDistributionStep(CubeSegment seg, String inputPath, String jobId) {
MapReduceExecutable rowkeyDistributionStep = new MapReduceExecutable();
rowkeyDistributionStep.setName(ExecutableConstants.STEP_NAME_GET_CUBOID_KEY_DISTRIBUTION);
StringBuilder cmd = new StringBuilder();
appendMapReduceParameters(cmd, seg);
appendExecCmdParameters(cmd, "input", inputPath);
- appendExecCmdParameters(cmd, "output", getRowkeyDistributionOutputPath(seg));
+ appendExecCmdParameters(cmd, "output", getRowkeyDistributionOutputPath(seg, jobId));
appendExecCmdParameters(cmd, "cubename", seg.getCubeInstance().getName());
appendExecCmdParameters(cmd, "jobname", "Kylin_Region_Splits_Calculator_" + seg.getCubeInstance().getName() + "_Step");
@@ -363,12 +363,12 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
return rowkeyDistributionStep;
}
- private HadoopShellExecutable createCreateHTableStep(CubeSegment seg) {
+ private HadoopShellExecutable createCreateHTableStep(CubeSegment seg, String jobId) {
HadoopShellExecutable createHtableStep = new HadoopShellExecutable();
createHtableStep.setName(ExecutableConstants.STEP_NAME_CREATE_HBASE_TABLE);
StringBuilder cmd = new StringBuilder();
appendExecCmdParameters(cmd, "cubename", seg.getCubeInstance().getName());
- appendExecCmdParameters(cmd, "input", getRowkeyDistributionOutputPath(seg) + "/part-r-00000");
+ appendExecCmdParameters(cmd, "input", getRowkeyDistributionOutputPath(seg, jobId) + "/part-r-00000");
appendExecCmdParameters(cmd, "htablename", seg.getStorageLocationIdentifier());
createHtableStep.setJobParams(cmd.toString());
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/9df88498/job/src/test/java/org/apache/kylin/job/cube/CubingJobBuilderTest.java
----------------------------------------------------------------------
diff --git a/job/src/test/java/org/apache/kylin/job/cube/CubingJobBuilderTest.java b/job/src/test/java/org/apache/kylin/job/cube/CubingJobBuilderTest.java
new file mode 100644
index 0000000..f8cbf10
--- /dev/null
+++ b/job/src/test/java/org/apache/kylin/job/cube/CubingJobBuilderTest.java
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kylin.job.cube;
+
+import org.apache.commons.cli.*;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.util.GenericOptionsParser;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.HBaseMetadataTestCase;
+import org.apache.kylin.cube.CubeManager;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.job.DeployUtil;
+import org.apache.kylin.job.common.MapReduceExecutable;
+import org.apache.kylin.job.constant.ExecutableConstants;
+import org.apache.kylin.job.engine.JobEngineConfig;
+import org.apache.kylin.job.hadoop.AbstractHadoopJob;
+import org.apache.kylin.metadata.model.SegmentStatusEnum;
+import org.junit.*;
+
+import java.io.IOException;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.TimeZone;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import static org.junit.Assert.*;
+
+/**
+ * Created by sunyerui on 15/8/31.
+ */
+public class CubingJobBuilderTest extends HBaseMetadataTestCase {
+
+ private JobEngineConfig jobEngineConfig;
+
+ private CubeManager cubeManager;
+
+ private static final Log logger = LogFactory.getLog(CubingJobBuilderTest.class);
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ staticCreateTestMetadata();
+ }
+
+ @Before
+ public void before() throws Exception {
+ DeployUtil.deployMetadata();
+
+ final KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
+ cubeManager = CubeManager.getInstance(kylinConfig);
+ jobEngineConfig = new JobEngineConfig(kylinConfig);
+ }
+
+ @AfterClass
+ public static void afterClass() {
+ staticCleanupTestMetadata();
+ }
+
+ private CubeSegment buildSegment() throws ParseException, IOException {
+ SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
+ f.setTimeZone(TimeZone.getTimeZone("GMT"));
+ // this cube's start date is 0, end date is 20501112000000
+ long date2 = f.parse("2022-01-01").getTime();
+
+ CubeSegment segment = cubeManager.appendSegments(cubeManager.getCube("test_kylin_cube_without_slr_empty"), date2);
+ // just to cheat the cubeManager.getBuildingSegments checking
+ segment.setStatus(SegmentStatusEnum.READY);
+
+ return segment;
+ }
+
+ private static class ParseOptionHelperJob extends AbstractHadoopJob {
+ @Override
+ public int run(String[] strings) throws Exception {
+ return 0;
+ }
+
+ public void parseOptionsForRowkeyDistributionStep(String arg) throws org.apache.commons.cli.ParseException, IOException {
+ Options options = new Options();
+ options.addOption(OPTION_INPUT_PATH);
+ options.addOption(OPTION_OUTPUT_PATH);
+ options.addOption(OPTION_JOB_NAME);
+ options.addOption(OPTION_CUBE_NAME);
+
+ GenericOptionsParser hadoopParser = new GenericOptionsParser(new Configuration(), arg.trim().split("\\s+"));
+ String[] toolArgs = hadoopParser.getRemainingArgs();
+ parseOptions(options, toolArgs);
+ }
+
+ public String getOutputPath() {
+ return getOptionValue(OPTION_OUTPUT_PATH);
+ }
+ }
+
+ private static final Pattern UUID_PATTERN = Pattern.compile(".*kylin-([0-9a-f]{8}(-[0-9a-f]{4}){3}-[0-9a-f]{12}).*rowkey_stats");
+ @Test
+ public void testBuildJob() throws ParseException, IOException, org.apache.commons.cli.ParseException {
+ CubeSegment segment = buildSegment();
+ CubingJobBuilder cubingJobBuilder = new CubingJobBuilder(jobEngineConfig);
+ CubingJob job = cubingJobBuilder.buildJob(segment);
+ assertNotNull(job);
+
+ // here should be more asserts for every step in building
+ // only check rowkey distribution step for now
+ MapReduceExecutable rowkeyDistributionStep = (MapReduceExecutable)job.getTaskByName(ExecutableConstants.STEP_NAME_GET_CUBOID_KEY_DISTRIBUTION);
+ assertNotNull(rowkeyDistributionStep);
+ String mrParams = rowkeyDistributionStep.getMapReduceParams();
+ assertNotNull(mrParams);
+ logger.info("mrParams: " + mrParams);
+ // parse output path and check
+ ParseOptionHelperJob parseHelper = new ParseOptionHelperJob();
+ parseHelper.parseOptionsForRowkeyDistributionStep(mrParams);
+ String outputPath = parseHelper.getOutputPath();
+ logger.info("output: " + outputPath);
+ Matcher m = UUID_PATTERN.matcher(outputPath);
+ assertTrue(m.find());
+ assertEquals(2, m.groupCount());
+ assertEquals(job.getId(), m.group(1));
+ }
+}
[2/2] incubator-kylin git commit: KYLIN-978 GarbageCollectionStep
dropped Hive Intermediate Table but didn't drop external hdfs path
Posted by sh...@apache.org.
KYLIN-978 GarbageCollectionStep dropped Hive Intermediate Table but didn't drop external hdfs path
Signed-off-by: shaofengshi <sh...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/bb1b1a72
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/bb1b1a72
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/bb1b1a72
Branch: refs/heads/KYLIN-968-978
Commit: bb1b1a725b81803d6b6e72735156d2d9cd067a6e
Parents: 9df8849
Author: sunyerui <su...@gmail.com>
Authored: Fri Sep 11 00:04:42 2015 +0800
Committer: shaofengshi <sh...@apache.org>
Committed: Fri Sep 11 11:07:26 2015 +0800
----------------------------------------------------------------------
.../apache/kylin/job/cube/CubingJobBuilder.java | 47 ++++++++++++--------
.../kylin/job/cube/GarbageCollectionStep.java | 38 ++++++++++++----
2 files changed, 58 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/bb1b1a72/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 ff79286..de75f7d 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
@@ -25,8 +25,6 @@ import java.util.List;
import java.util.TimeZone;
import org.apache.commons.lang3.StringUtils;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
import org.apache.kylin.common.util.HadoopUtil;
import org.apache.kylin.common.util.Pair;
import org.apache.kylin.cube.CubeSegment;
@@ -65,9 +63,11 @@ 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();
// cubing
- Pair<AbstractExecutable, AbstractExecutable> twoSteps = addCubingSteps(seg, cuboidRootPath, result);
+ Pair<AbstractExecutable, AbstractExecutable> twoSteps = addCubingSteps(seg, cuboidRootPath, result, toDeletePathsOnHadoopCluster);
String intermediateHiveTableStepId = twoSteps.getFirst().getId();
String baseCuboidStepId = twoSteps.getSecond().getId();
@@ -79,7 +79,8 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
final CubeJoinedFlatTableDesc intermediateTableDesc = new CubeJoinedFlatTableDesc(seg.getCubeDesc(), seg);
final String hiveIntermediateTable = this.getIntermediateHiveTableName(intermediateTableDesc, jobId);
- result.addTask(createGarbageCollectionStep(seg, null, hiveIntermediateTable, null));
+ toDeletePathsOnHBaseCluster.add(getJobWorkingDir(jobId));
+ result.addTask(createGarbageCollectionStep(seg, null, hiveIntermediateTable, toDeletePathsOnHadoopCluster, toDeletePathsOnHBaseCluster));
return result;
}
@@ -92,9 +93,14 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
final String jobId = result.getId();
final String appendRootPath = getJobWorkingDir(jobId) + "/" + appendSegment.getCubeInstance().getName() + "/append_cuboid/";
final String mergedRootPath = getJobWorkingDir(jobId) + "/" + appendSegment.getCubeInstance().getName() + "/cuboid/";
+ 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();
// cubing the incremental segment
- Pair<AbstractExecutable, AbstractExecutable> twoSteps = addCubingSteps(appendSegment, appendRootPath, result);
+ Pair<AbstractExecutable, AbstractExecutable> twoSteps = addCubingSteps(appendSegment, appendRootPath, result, toDeletePathsOnHadoopCluster);
final String intermediateHiveTableStepId = twoSteps.getFirst().getId();
final String baseCuboidStepId = twoSteps.getSecond().getId();
@@ -103,10 +109,7 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
List<CubeSegment> mergingSegments = mergeSegment.getCubeInstance().getMergingSegments(mergeSegment);
Preconditions.checkState(mergingSegments.size() > 1, "there should be more than 2 segments to merge");
- List<String> mergingSegmentIds = Lists.newArrayList();
- List<String> mergingCuboidPaths = Lists.newArrayList();
- List<String> mergingHTables = Lists.newArrayList();
- List<String> toDeletePaths = Lists.newArrayList();
+
for (CubeSegment merging : mergingSegments) {
mergingSegmentIds.add(merging.getUuid());
mergingHTables.add(merging.getStorageLocationIdentifier());
@@ -115,7 +118,7 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
} else {
mergingCuboidPaths.add(getPathToMerge(merging));
}
- toDeletePaths.add(getJobWorkingDir(merging.getLastBuildJobID()));
+ toDeletePathsOnHadoopCluster.add(getJobWorkingDir(merging.getLastBuildJobID()));
}
// merge cuboid
@@ -126,7 +129,8 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
// update cube info
result.addTask(createUpdateCubeInfoAfterMergeStep(mergeSegment, mergingSegmentIds, convertCuboidToHfileStep.getId(), jobId));
- result.addTask(createGarbageCollectionStep(mergeSegment, mergingHTables, null, toDeletePaths));
+ toDeletePathsOnHBaseCluster.add(getJobWorkingDir(jobId));
+ result.addTask(createGarbageCollectionStep(mergeSegment, mergingHTables, null, toDeletePathsOnHadoopCluster, toDeletePathsOnHBaseCluster));
return result;
}
@@ -143,12 +147,14 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
List<String> mergingSegmentIds = Lists.newArrayList();
List<String> mergingCuboidPaths = Lists.newArrayList();
List<String> mergingHTables = Lists.newArrayList();
- List<String> toDeletePaths = Lists.newArrayList();
+ final List<String> toDeletePathsOnHadoopCluster = Lists.newArrayList();
+ final List<String> toDeletePathsOnHBaseCluster = Lists.newArrayList();
+
for (CubeSegment merging : mergingSegments) {
mergingSegmentIds.add(merging.getUuid());
mergingCuboidPaths.add(getPathToMerge(merging));
mergingHTables.add(merging.getStorageLocationIdentifier());
- toDeletePaths.add(getJobWorkingDir(merging.getLastBuildJobID()));
+ toDeletePathsOnHadoopCluster.add(getJobWorkingDir(merging.getLastBuildJobID()));
}
// merge cuboid
@@ -159,7 +165,8 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
// update cube info
result.addTask(createUpdateCubeInfoAfterMergeStep(seg, mergingSegmentIds, convertCuboidToHfileStep.getId(), jobId));
- result.addTask(createGarbageCollectionStep(seg, mergingHTables, null, toDeletePaths));
+ toDeletePathsOnHBaseCluster.add(getJobWorkingDir(jobId));
+ result.addTask(createGarbageCollectionStep(seg, mergingHTables, null, toDeletePathsOnHadoopCluster, toDeletePathsOnHBaseCluster));
return result;
}
@@ -171,7 +178,7 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
result.addTask(createMergeCuboidDataStep(seg, formattedPath, mergedCuboidPath));
}
- Pair<AbstractExecutable, AbstractExecutable> addCubingSteps(CubeSegment seg, String cuboidRootPath, CubingJob result) {
+ Pair<AbstractExecutable, AbstractExecutable> addCubingSteps(CubeSegment seg, String cuboidRootPath, CubingJob result, List<String> toDeletePaths) {
final int groupRowkeyColumnsCount = seg.getCubeDesc().getRowkey().getNCuboidBuildLevels();
final int totalRowkeyColumnsCount = seg.getCubeDesc().getRowkey().getRowKeyColumns().length;
@@ -199,6 +206,9 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
result.addTask(createNDimensionCuboidStep(seg, cuboidOutputTempPath, dimNum, totalRowkeyColumnsCount));
}
+ toDeletePaths.add(intermediateHiveTableLocation);
+ toDeletePaths.add(factDistinctColumnsPath);
+
return new Pair<AbstractExecutable, AbstractExecutable>(intermediateHiveTableStep, baseCuboidStep);
}
@@ -266,7 +276,7 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
}
private String getRowkeyDistributionOutputPath(CubeSegment seg, String jobId) {
- return getJobWorkingDir(jobId) + "/" + seg.getCubeInstance().getName() + "/rowkey_stats";
+ return HadoopUtil.makeQualifiedPathInHBaseCluster(getJobWorkingDir(jobId) + "/" + seg.getCubeInstance().getName() + "/rowkey_stats");
}
private String getFactDistinctColumnsPath(CubeSegment seg, String jobUuid) {
@@ -460,12 +470,13 @@ public final class CubingJobBuilder extends AbstractJobBuilder {
return result;
}
- private GarbageCollectionStep createGarbageCollectionStep(CubeSegment seg, List<String> oldHtables, String hiveIntermediateTable, List<String> oldHdsfPaths) {
+ private GarbageCollectionStep createGarbageCollectionStep(CubeSegment seg, List<String> oldHtables, String hiveIntermediateTable, List<String> oldHdsfPaths, List<String> oldHdfsPathsOnHBaseCluster) {
GarbageCollectionStep result = new GarbageCollectionStep();
result.setName(ExecutableConstants.STEP_NAME_GARBAGE_COLLECTION);
result.setOldHTables(oldHtables);
result.setOldHiveTable(hiveIntermediateTable);
- result.setOldHdsfPaths(oldHdsfPaths);
+ result.setOldHdfsPaths(oldHdsfPaths);
+ result.setOldHdfsPathsOnHBaseCluster(oldHdfsPathsOnHBaseCluster);
return result;
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/bb1b1a72/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 72cad96..641454c 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,6 +55,8 @@ 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;
@@ -69,8 +71,9 @@ public class GarbageCollectionStep extends AbstractExecutable {
try {
dropHBaseTable(context);
- dropHdfsPath(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());
@@ -131,13 +134,11 @@ public class GarbageCollectionStep extends AbstractExecutable {
}
}
}
-
- private void dropHdfsPath(ExecutableContext context) throws IOException {
- List<String> oldHdfsPaths = this.getOldHdsfPaths();
+ private void dropHdfsPathOnCluster(List<String> oldHdfsPaths, FileSystem fileSystem) throws IOException {
if (oldHdfsPaths != null && oldHdfsPaths.size() > 0) {
- Configuration hconf = HadoopUtil.getCurrentConfiguration();
- FileSystem fileSystem = FileSystem.get(hconf);
+ logger.debug("Drop HDFS path on FileSystem: " + fileSystem.getUri());
+ output.append("Drop HDFS path on FileSystem: \"" + fileSystem.getUri() + "\" \n");
for (String path : oldHdfsPaths) {
if (path.endsWith("*"))
path = path.substring(0, path.length() - 1);
@@ -152,10 +153,21 @@ public class GarbageCollectionStep extends AbstractExecutable {
output.append("HDFS path not exists: \"" + path + "\" \n");
}
}
-
}
}
+ private void dropHdfsPath(ExecutableContext context) throws IOException {
+ 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);
+ }
+
public void setOldHTables(List<String> tables) {
setArrayParam(OLD_HTABLES, tables);
}
@@ -164,14 +176,22 @@ public class GarbageCollectionStep extends AbstractExecutable {
return getArrayParam(OLD_HTABLES);
}
- public void setOldHdsfPaths(List<String> paths) {
+ public void setOldHdfsPaths(List<String> paths) {
setArrayParam(OLD_HDFS_PATHS, paths);
}
- private List<String> getOldHdsfPaths() {
+ private List<String> getOldHdfsPaths() {
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, ","));
}