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 2018/12/10 03:41:29 UTC
[kylin] 05/06: KYLIN-3626 Allow customization for storage path
This is an automated email from the ASF dual-hosted git repository.
shaofengshi pushed a commit to branch kylin-on-parquet
in repository https://gitbox.apache.org/repos/asf/kylin.git
commit 8bbb77747b61a70676b0734e27ac25e288aa6065
Author: chao long <wa...@qq.com>
AuthorDate: Fri Nov 30 08:21:09 2018 +0800
KYLIN-3626 Allow customization for storage path
---
.../org/apache/kylin/common/KylinConfigBase.java | 4 ++
.../storage/path/DefaultStoragePathBuilder.java | 49 ++++++++++++++++
.../kylin/storage/path/IStoragePathBuilder.java | 31 +++++++++++
.../kylin/storage/path/S3ReversePathBuilder.java | 65 ++++++++++++++++++++++
.../apache/kylin/engine/mr/JobBuilderSupport.java | 47 +++++++---------
.../apache/kylin/rest/job/StorageCleanupJob.java | 13 +++--
.../apache/kylin/source/hive/HiveInputBase.java | 11 ++--
.../org/apache/kylin/source/hive/HiveMRInput.java | 10 +++-
.../apache/kylin/source/hive/HiveSparkInput.java | 11 ++--
.../apache/kylin/source/hive/HiveMRInputTest.java | 5 +-
.../apache/kylin/source/jdbc/JdbcHiveMRInput.java | 2 +-
.../apache/kylin/source/kafka/KafkaMRInput.java | 11 +++-
.../apache/kylin/source/kafka/KafkaSparkInput.java | 7 ++-
.../storage/hbase/lookup/HBaseLookupMRSteps.java | 6 +-
.../hbase/steps/HDFSPathGarbageCollectionStep.java | 7 ++-
.../kylin/storage/hbase/util/CubeMigrationCLI.java | 9 ++-
.../storage/hbase/util/StorageCleanupJob.java | 8 ++-
.../storage/parquet/steps/ParquetMRSteps.java | 2 +-
.../storage/parquet/steps/ParquetSparkSteps.java | 2 +-
.../org/apache/kylin/tool/CubeMigrationCLI.java | 10 +++-
20 files changed, 245 insertions(+), 65 deletions(-)
diff --git a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
index e3e3e03..2633ddf 100644
--- a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
+++ b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
@@ -1863,4 +1863,8 @@ abstract public class KylinConfigBase implements Serializable {
public String getJdbcSourceAdaptor() {
return getOptional("kylin.source.jdbc.adaptor");
}
+
+ public String getStorageSystemPathBuilderClz() {
+ return getOptional("storage.path.builder", "org.apache.kylin.storage.path.DefaultStoragePathBuilder");
+ }
}
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/path/DefaultStoragePathBuilder.java b/core-storage/src/main/java/org/apache/kylin/storage/path/DefaultStoragePathBuilder.java
new file mode 100644
index 0000000..1e50ba0
--- /dev/null
+++ b/core-storage/src/main/java/org/apache/kylin/storage/path/DefaultStoragePathBuilder.java
@@ -0,0 +1,49 @@
+/*
+ * 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.storage.path;
+
+import org.apache.kylin.cube.CubeSegment;
+
+public class DefaultStoragePathBuilder implements IStoragePathBuilder {
+
+ @Override
+ public String getJobWorkingDir(String workingDir, String jobId) {
+ if (!workingDir.endsWith(SLASH)) {
+ workingDir = workingDir + SLASH;
+ }
+
+ return workingDir + "kylin-" + jobId;
+ }
+
+ @Override
+ public String getJobRealizationRootPath(CubeSegment cubeSegment, String jobId) {
+ String jobWorkingDir = getJobWorkingDir(cubeSegment.getConfig().getHdfsWorkingDirectory(), jobId);
+
+ return jobWorkingDir + SLASH + cubeSegment.getRealization().getName();
+ }
+
+ @Override
+ public String getRealizationFinalDataPath(CubeSegment cubeSegment) {
+ String workingDir = cubeSegment.getConfig().getHdfsWorkingDirectory();
+ String cubeName = cubeSegment.getRealization().getName();
+ String segName = cubeSegment.getName();
+
+ return workingDir + SLASH + cubeName + SLASH + segName;
+ }
+}
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/path/IStoragePathBuilder.java b/core-storage/src/main/java/org/apache/kylin/storage/path/IStoragePathBuilder.java
new file mode 100644
index 0000000..7a70f98
--- /dev/null
+++ b/core-storage/src/main/java/org/apache/kylin/storage/path/IStoragePathBuilder.java
@@ -0,0 +1,31 @@
+/*
+ * 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.storage.path;
+
+import org.apache.kylin.cube.CubeSegment;
+
+public interface IStoragePathBuilder {
+ public static final String SLASH = "/";
+
+ public String getJobWorkingDir(String workingDir, String jobId);
+
+ public String getJobRealizationRootPath(CubeSegment cubeSegment, String jobId);
+
+ public String getRealizationFinalDataPath(CubeSegment cubeSegment);
+}
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/path/S3ReversePathBuilder.java b/core-storage/src/main/java/org/apache/kylin/storage/path/S3ReversePathBuilder.java
new file mode 100644
index 0000000..5b01ad3
--- /dev/null
+++ b/core-storage/src/main/java/org/apache/kylin/storage/path/S3ReversePathBuilder.java
@@ -0,0 +1,65 @@
+/*
+ * 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.storage.path;
+
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.kylin.cube.CubeSegment;
+
+public class S3ReversePathBuilder implements IStoragePathBuilder {
+
+ protected String prefix;
+ protected String bucketName;
+
+ protected String getReverseMetaDir(String workingDir) {
+ int prefixIndex = workingDir.indexOf("//") + 2;
+ int dirIndex = workingDir.indexOf(SLASH, prefixIndex);
+
+ this.prefix = workingDir.substring(0, prefixIndex);
+ this.bucketName = workingDir.substring(prefixIndex, dirIndex);
+
+ String[] dirs = workingDir.substring(dirIndex + 1).split(SLASH);
+ ArrayUtils.reverse(dirs);
+
+ return StringUtils.join(dirs, SLASH);
+ }
+
+ @Override
+ public String getJobWorkingDir(String workingDir, String jobId) {
+ String reverseMetaDir = getReverseMetaDir(workingDir);
+
+ return this.prefix + this.bucketName + SLASH + "kylin-" + jobId + SLASH + reverseMetaDir;
+ }
+
+ @Override
+ public String getJobRealizationRootPath(CubeSegment cubeSegment, String jobId) {
+ String jobWorkingDir = getJobWorkingDir(cubeSegment.getConfig().getHdfsWorkingDirectory(), jobId);
+
+ return jobWorkingDir + SLASH + cubeSegment.getRealization().getName();
+ }
+
+ @Override
+ public String getRealizationFinalDataPath(CubeSegment cubeSegment) {
+ String reverseMetaDir = getReverseMetaDir(cubeSegment.getConfig().getHdfsWorkingDirectory());
+ String cubeName = cubeSegment.getRealization().getName();
+ String segName = cubeSegment.getName();
+
+ return this.prefix + this.bucketName + SLASH + segName + SLASH + cubeName + SLASH + reverseMetaDir;
+ }
+}
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/JobBuilderSupport.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/JobBuilderSupport.java
index fd11c99..8525dcc 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/JobBuilderSupport.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/JobBuilderSupport.java
@@ -27,6 +27,7 @@ import java.util.regex.Pattern;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.StorageURL;
+import org.apache.kylin.common.util.ClassUtil;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.CuboidModeEnum;
import org.apache.kylin.cube.model.CubeDesc;
@@ -50,6 +51,9 @@ import org.apache.kylin.job.engine.JobEngineConfig;
import org.apache.kylin.metadata.model.TblColRef;
import com.google.common.base.Preconditions;
+import org.apache.kylin.storage.path.IStoragePathBuilder;
+
+import static org.apache.kylin.storage.path.IStoragePathBuilder.SLASH;
/**
* Hold reusable steps for builders.
@@ -59,6 +63,7 @@ public class JobBuilderSupport {
final protected JobEngineConfig config;
final protected CubeSegment seg;
final protected String submitter;
+ final protected IStoragePathBuilder storagePathBuilder;
final public static String LayeredCuboidFolderPrefix = "level_";
@@ -72,6 +77,8 @@ public class JobBuilderSupport {
this.config = new JobEngineConfig(seg.getConfig());
this.seg = seg;
this.submitter = submitter;
+ String pathBuilderClz = seg.getConfig().getStorageSystemPathBuilderClz();
+ this.storagePathBuilder = (IStoragePathBuilder)ClassUtil.newInstance(pathBuilderClz);
}
public MapReduceExecutable createFactDistinctColumnsStep(String jobId) {
@@ -272,11 +279,11 @@ public class JobBuilderSupport {
// ============================================================================
public String getJobWorkingDir(String jobId) {
- return getJobWorkingDir(config, jobId);
+ return storagePathBuilder.getJobWorkingDir(config.getConfig().getHdfsWorkingDirectory(), jobId);
}
public String getRealizationRootPath(String jobId) {
- return getJobWorkingDir(jobId) + "/" + seg.getRealization().getName();
+ return storagePathBuilder.getJobRealizationRootPath(seg, jobId);
}
public String getCuboidRootPath(String jobId) {
@@ -311,7 +318,7 @@ public class JobBuilderSupport {
}
public String getStatisticsPath(String jobId) {
- return getRealizationRootPath(jobId) + "/fact_distinct_columns/" + BatchConstants.CFG_OUTPUT_STATISTICS;
+ return getFactDistinctColumnsPath(jobId) + SLASH + BatchConstants.CFG_OUTPUT_STATISTICS;
}
public String getShrunkenDictionaryPath(String jobId) {
@@ -346,29 +353,24 @@ public class JobBuilderSupport {
return getRealizationRootPath(jobId) + "/counter";
}
- public String getParquetOutputPath(String jobId) {
- return getRealizationRootPath(jobId) + "/parquet/";
+ public String getParquetOutputPath() {
+ return storagePathBuilder.getRealizationFinalDataPath(seg) + "/";
}
- public String getParquetOutputPath() {
- return getParquetOutputPath(seg.getLastBuildJobID());
+ public String getDumpMetadataPath(String jobId) {
+ return getRealizationRootPath(jobId) + "/metadata";
+ }
+
+ public String getSegmentMetadataUrl(KylinConfig kylinConfig, String jobId) {
+ Map<String, String> param = new HashMap<>();
+ param.put("path", getDumpMetadataPath(jobId));
+ return new StorageURL(kylinConfig.getMetadataUrl().getIdentifier(), "hdfs", param).toString();
}
// ============================================================================
// static methods also shared by other job flow participant
// ----------------------------------------------------------------------------
- public static String getJobWorkingDir(JobEngineConfig conf, String jobId) {
- return getJobWorkingDir(conf.getHdfsWorkingDirectory(), jobId);
- }
-
- public static String getJobWorkingDir(String hdfsDir, String jobId) {
- if (!hdfsDir.endsWith("/")) {
- hdfsDir = hdfsDir + "/";
- }
- return hdfsDir + "kylin-" + jobId;
- }
-
public static StringBuilder appendExecCmdParameters(StringBuilder buf, String paraName, String paraValue) {
return buf.append(" -").append(paraName).append(" ").append(paraValue);
}
@@ -389,10 +391,6 @@ public class JobBuilderSupport {
return cuboidRootPath + PathNameCuboidInMem;
}
- public String getDumpMetadataPath(String jobId) {
- return getRealizationRootPath(jobId) + "/metadata";
- }
-
public static String extractJobIDFromPath(String path) {
Matcher matcher = JOB_NAME_PATTERN.matcher(path);
// check the first occurrence
@@ -403,9 +401,4 @@ public class JobBuilderSupport {
}
}
- public String getSegmentMetadataUrl(KylinConfig kylinConfig, String jobId) {
- Map<String, String> param = new HashMap<>();
- param.put("path", getDumpMetadataPath(jobId));
- return new StorageURL(kylinConfig.getMetadataUrl().getIdentifier(), "hdfs", param).toString();
- }
}
diff --git a/server-base/src/main/java/org/apache/kylin/rest/job/StorageCleanupJob.java b/server-base/src/main/java/org/apache/kylin/rest/job/StorageCleanupJob.java
index b73e916..6900ca7 100755
--- a/server-base/src/main/java/org/apache/kylin/rest/job/StorageCleanupJob.java
+++ b/server-base/src/main/java/org/apache/kylin/rest/job/StorageCleanupJob.java
@@ -41,6 +41,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.util.AbstractApplication;
+import org.apache.kylin.common.util.ClassUtil;
import org.apache.kylin.common.util.CliCommandExecutor;
import org.apache.kylin.common.util.HadoopUtil;
import org.apache.kylin.common.util.HiveCmdBuilder;
@@ -49,7 +50,6 @@ import org.apache.kylin.common.util.Pair;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.engine.mr.JobBuilderSupport;
import org.apache.kylin.job.engine.JobEngineConfig;
import org.apache.kylin.job.execution.AbstractExecutable;
import org.apache.kylin.job.execution.ExecutableManager;
@@ -57,6 +57,7 @@ import org.apache.kylin.job.execution.ExecutableState;
import org.apache.kylin.metadata.MetadataConstants;
import org.apache.kylin.source.ISourceMetadataExplorer;
import org.apache.kylin.source.SourceManager;
+import org.apache.kylin.storage.path.IStoragePathBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -81,7 +82,8 @@ public class StorageCleanupJob extends AbstractApplication {
final protected FileSystem hbaseFs;
final protected FileSystem defaultFs;
final protected ExecutableManager executableManager;
-
+ final protected IStoragePathBuilder pathBuilder;
+
protected boolean delete = false;
protected boolean force = false;
@@ -100,6 +102,7 @@ public class StorageCleanupJob extends AbstractApplication {
this.defaultFs = defaultFs;
this.hbaseFs = hbaseFs;
this.executableManager = ExecutableManager.getInstance(config);
+ this.pathBuilder = (IStoragePathBuilder)ClassUtil.newInstance(config.getStorageSystemPathBuilderClz());
}
public void setDelete(boolean delete) {
@@ -248,7 +251,7 @@ public class StorageCleanupJob extends AbstractApplication {
for (String jobId : allJobs) {
final ExecutableState state = executableManager.getOutput(jobId).getState();
if (!state.isFinalState()) {
- String path = JobBuilderSupport.getJobWorkingDir(engineConfig.getHdfsWorkingDirectory(), jobId);
+ String path = pathBuilder.getJobWorkingDir(engineConfig.getHdfsWorkingDirectory(), jobId);
allHdfsPathsNeedToBeDeleted.remove(path);
logger.info("Skip " + path + " from deletion list, as the path belongs to job " + jobId
+ " with status " + state);
@@ -260,7 +263,7 @@ public class StorageCleanupJob extends AbstractApplication {
for (CubeSegment seg : cube.getSegments()) {
String jobUuid = seg.getLastBuildJobID();
if (jobUuid != null && jobUuid.equals("") == false) {
- String path = JobBuilderSupport.getJobWorkingDir(engineConfig.getHdfsWorkingDirectory(), jobUuid);
+ String path = pathBuilder.getJobWorkingDir(engineConfig.getHdfsWorkingDirectory(), jobUuid);
allHdfsPathsNeedToBeDeleted.remove(path);
logger.info("Skip " + path + " from deletion list, as the path belongs to segment " + seg
+ " of cube " + cube.getName());
@@ -415,7 +418,7 @@ public class StorageCleanupJob extends AbstractApplication {
String segmentId = uuid.replace("_", "-");
if (segmentId2JobId.containsKey(segmentId)) {
- String path = JobBuilderSupport.getJobWorkingDir(engineConfig.getHdfsWorkingDirectory(),
+ String path = pathBuilder.getJobWorkingDir(engineConfig.getHdfsWorkingDirectory(),
segmentId2JobId.get(segmentId)) + "/" + tableToDelete;
Path externalDataPath = new Path(path);
if (defaultFs.exists(externalDataPath)) {
diff --git a/source-hive/src/main/java/org/apache/kylin/source/hive/HiveInputBase.java b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveInputBase.java
index 2c998df..51954d8 100644
--- a/source-hive/src/main/java/org/apache/kylin/source/hive/HiveInputBase.java
+++ b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveInputBase.java
@@ -29,7 +29,6 @@ import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.util.HadoopUtil;
import org.apache.kylin.common.util.HiveCmdBuilder;
import org.apache.kylin.cube.model.CubeDesc;
-import org.apache.kylin.engine.mr.JobBuilderSupport;
import org.apache.kylin.engine.mr.steps.CubingExecutableUtil;
import org.apache.kylin.job.JoinedFlatTable;
import org.apache.kylin.job.common.ShellExecutable;
@@ -41,6 +40,7 @@ import org.apache.kylin.metadata.model.DataModelDesc;
import org.apache.kylin.metadata.model.IJoinedFlatTableDesc;
import org.apache.kylin.metadata.model.JoinTableDesc;
import org.apache.kylin.metadata.model.TableDesc;
+import org.apache.kylin.storage.path.IStoragePathBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -58,11 +58,11 @@ public class HiveInputBase {
return String.format(Locale.ROOT, "%s.%s", database, tableName).toUpperCase(Locale.ROOT);
}
- protected void addStepPhase1_DoCreateFlatTable(DefaultChainedExecutable jobFlow, String hdfsWorkingDir,
+ protected void addStepPhase1_DoCreateFlatTable(DefaultChainedExecutable jobFlow, String hdfsWorkingDir, IStoragePathBuilder pathBuilder,
IJoinedFlatTableDesc flatTableDesc, String flatTableDatabase) {
final String cubeName = CubingExecutableUtil.getCubeName(jobFlow.getParams());
final String hiveInitStatements = JoinedFlatTable.generateHiveInitStatements(flatTableDatabase);
- final String jobWorkingDir = getJobWorkingDir(jobFlow, hdfsWorkingDir);
+ final String jobWorkingDir = getJobWorkingDir(jobFlow, hdfsWorkingDir, pathBuilder);
jobFlow.addTask(createFlatHiveTableStep(hiveInitStatements, jobWorkingDir, cubeName, flatTableDesc));
}
@@ -142,9 +142,10 @@ public class HiveInputBase {
return createIntermediateTableHql.toString();
}
- protected static String getJobWorkingDir(DefaultChainedExecutable jobFlow, String hdfsWorkingDir) {
+ protected static String getJobWorkingDir(DefaultChainedExecutable jobFlow, String hdfsWorkingDir, IStoragePathBuilder pathBuilder) {
+
+ String jobWorkingDir = pathBuilder.getJobWorkingDir(hdfsWorkingDir, jobFlow.getId());
- String jobWorkingDir = JobBuilderSupport.getJobWorkingDir(hdfsWorkingDir, jobFlow.getId());
if (KylinConfig.getInstanceFromEnv().getHiveTableDirCreateFirst()) {
// Create work dir to avoid hive create it,
// the difference is that the owners are different.
diff --git a/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
index d6b85ed..3c9434b 100644
--- a/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
+++ b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
@@ -29,6 +29,7 @@ import org.apache.hive.hcatalog.data.HCatRecord;
import org.apache.hive.hcatalog.mapreduce.HCatInputFormat;
import org.apache.hive.hcatalog.mapreduce.HCatSplit;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.ClassUtil;
import org.apache.kylin.common.util.HadoopUtil;
import org.apache.kylin.common.util.StringUtil;
import org.apache.kylin.cube.CubeInstance;
@@ -43,6 +44,7 @@ import org.apache.kylin.job.execution.ExecutableManager;
import org.apache.kylin.metadata.model.IJoinedFlatTableDesc;
import org.apache.kylin.metadata.model.ISegment;
import org.apache.kylin.metadata.model.TableDesc;
+import org.apache.kylin.storage.path.IStoragePathBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -116,6 +118,7 @@ public class HiveMRInput extends HiveInputBase implements IMRInput {
final protected IJoinedFlatTableDesc flatDesc;
final protected String flatTableDatabase;
final protected String hdfsWorkingDir;
+ final protected IStoragePathBuilder pathBuilder;
List<String> hiveViewIntermediateTables = Lists.newArrayList();
@@ -124,6 +127,7 @@ public class HiveMRInput extends HiveInputBase implements IMRInput {
this.flatDesc = flatDesc;
this.flatTableDatabase = config.getHiveDatabaseForIntermediateTable();
this.hdfsWorkingDir = config.getHdfsWorkingDirectory();
+ this.pathBuilder = (IStoragePathBuilder)ClassUtil.newInstance(config.getStorageSystemPathBuilderClz());
}
@Override
@@ -150,14 +154,14 @@ public class HiveMRInput extends HiveInputBase implements IMRInput {
protected void addStepPhase1_DoCreateFlatTable(DefaultChainedExecutable jobFlow) {
final String cubeName = CubingExecutableUtil.getCubeName(jobFlow.getParams());
final String hiveInitStatements = JoinedFlatTable.generateHiveInitStatements(flatTableDatabase);
- final String jobWorkingDir = getJobWorkingDir(jobFlow, hdfsWorkingDir);
+ final String jobWorkingDir = getJobWorkingDir(jobFlow, hdfsWorkingDir, pathBuilder);
jobFlow.addTask(createFlatHiveTableStep(hiveInitStatements, jobWorkingDir, cubeName, flatDesc));
}
protected void addStepPhase1_DoMaterializeLookupTable(DefaultChainedExecutable jobFlow) {
final String hiveInitStatements = JoinedFlatTable.generateHiveInitStatements(flatTableDatabase);
- final String jobWorkingDir = getJobWorkingDir(jobFlow, hdfsWorkingDir);
+ final String jobWorkingDir = getJobWorkingDir(jobFlow, hdfsWorkingDir, pathBuilder);
AbstractExecutable task = createLookupHiveViewMaterializationStep(hiveInitStatements, jobWorkingDir,
flatDesc, hiveViewIntermediateTables, jobFlow.getId());
@@ -168,7 +172,7 @@ public class HiveMRInput extends HiveInputBase implements IMRInput {
@Override
public void addStepPhase4_Cleanup(DefaultChainedExecutable jobFlow) {
- final String jobWorkingDir = getJobWorkingDir(jobFlow, hdfsWorkingDir);
+ final String jobWorkingDir = getJobWorkingDir(jobFlow, hdfsWorkingDir, pathBuilder);
org.apache.kylin.source.hive.GarbageCollectionStep step = new org.apache.kylin.source.hive.GarbageCollectionStep();
step.setName(ExecutableConstants.STEP_NAME_HIVE_CLEANUP);
diff --git a/source-hive/src/main/java/org/apache/kylin/source/hive/HiveSparkInput.java b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveSparkInput.java
index d710db7..031548c8 100644
--- a/source-hive/src/main/java/org/apache/kylin/source/hive/HiveSparkInput.java
+++ b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveSparkInput.java
@@ -22,6 +22,7 @@ import java.util.Collections;
import java.util.List;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.ClassUtil;
import org.apache.kylin.common.util.StringUtil;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
@@ -33,6 +34,7 @@ import org.apache.kylin.job.execution.AbstractExecutable;
import org.apache.kylin.job.execution.DefaultChainedExecutable;
import org.apache.kylin.metadata.model.IJoinedFlatTableDesc;
import org.apache.kylin.metadata.model.ISegment;
+import org.apache.kylin.storage.path.IStoragePathBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -63,7 +65,7 @@ public class HiveSparkInput extends HiveInputBase implements ISparkInput {
final protected IJoinedFlatTableDesc flatDesc;
final protected String flatTableDatabase;
final protected String hdfsWorkingDir;
-
+ final protected IStoragePathBuilder pathBuilder;
List<String> hiveViewIntermediateTables = Lists.newArrayList();
public BatchCubingInputSide(IJoinedFlatTableDesc flatDesc) {
@@ -71,6 +73,7 @@ public class HiveSparkInput extends HiveInputBase implements ISparkInput {
this.flatDesc = flatDesc;
this.flatTableDatabase = config.getHiveDatabaseForIntermediateTable();
this.hdfsWorkingDir = config.getHdfsWorkingDirectory();
+ this.pathBuilder = (IStoragePathBuilder)ClassUtil.newInstance(config.getStorageSystemPathBuilderClz());
}
@Override
@@ -81,7 +84,7 @@ public class HiveSparkInput extends HiveInputBase implements ISparkInput {
final String hiveInitStatements = JoinedFlatTable.generateHiveInitStatements(flatTableDatabase);
// create flat table first
- addStepPhase1_DoCreateFlatTable(jobFlow, hdfsWorkingDir, flatDesc, flatTableDatabase);
+ addStepPhase1_DoCreateFlatTable(jobFlow, hdfsWorkingDir, pathBuilder, flatDesc, flatTableDatabase);
// then count and redistribute
if (cubeConfig.isHiveRedistributeEnabled()) {
@@ -95,7 +98,7 @@ public class HiveSparkInput extends HiveInputBase implements ISparkInput {
protected void addStepPhase1_DoMaterializeLookupTable(DefaultChainedExecutable jobFlow) {
final String hiveInitStatements = JoinedFlatTable.generateHiveInitStatements(flatTableDatabase);
- final String jobWorkingDir = getJobWorkingDir(jobFlow, hdfsWorkingDir);
+ final String jobWorkingDir = getJobWorkingDir(jobFlow, hdfsWorkingDir, pathBuilder);
AbstractExecutable task = createLookupHiveViewMaterializationStep(hiveInitStatements, jobWorkingDir,
flatDesc, hiveViewIntermediateTables, jobFlow.getId());
@@ -106,7 +109,7 @@ public class HiveSparkInput extends HiveInputBase implements ISparkInput {
@Override
public void addStepPhase4_Cleanup(DefaultChainedExecutable jobFlow) {
- final String jobWorkingDir = getJobWorkingDir(jobFlow, hdfsWorkingDir);
+ final String jobWorkingDir = getJobWorkingDir(jobFlow, hdfsWorkingDir, pathBuilder);
GarbageCollectionStep step = new GarbageCollectionStep();
step.setName(ExecutableConstants.STEP_NAME_HIVE_CLEANUP);
diff --git a/source-hive/src/test/java/org/apache/kylin/source/hive/HiveMRInputTest.java b/source-hive/src/test/java/org/apache/kylin/source/hive/HiveMRInputTest.java
index 384c2b8..1bcd9ce 100644
--- a/source-hive/src/test/java/org/apache/kylin/source/hive/HiveMRInputTest.java
+++ b/source-hive/src/test/java/org/apache/kylin/source/hive/HiveMRInputTest.java
@@ -28,8 +28,10 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.KylinConfig.SetAndUnsetThreadLocalConfig;
+import org.apache.kylin.common.util.ClassUtil;
import org.apache.kylin.common.util.RandomUtil;
import org.apache.kylin.job.execution.DefaultChainedExecutable;
+import org.apache.kylin.storage.path.IStoragePathBuilder;
import org.junit.Assert;
import org.junit.Test;
@@ -43,10 +45,11 @@ public class HiveMRInputTest {
try (SetAndUnsetThreadLocalConfig autoUnset = KylinConfig.setAndUnsetThreadLocalConfig(kylinConfig)) {
when(kylinConfig.getHiveTableDirCreateFirst()).thenReturn(true);
when(kylinConfig.getHdfsWorkingDirectory()).thenReturn("/tmp/kylin/");
+ when(kylinConfig.getStorageSystemPathBuilderClz()).thenReturn("org.apache.kylin.storage.path.DefaultStoragePathBuilder");
DefaultChainedExecutable defaultChainedExecutable = mock(DefaultChainedExecutable.class);
defaultChainedExecutable.setId(RandomUtil.randomUUID().toString());
- String jobWorkingDir = HiveInputBase.getJobWorkingDir(defaultChainedExecutable, KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory());
+ String jobWorkingDir = HiveInputBase.getJobWorkingDir(defaultChainedExecutable, KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory(), (IStoragePathBuilder) ClassUtil.newInstance(KylinConfig.getInstanceFromEnv().getStorageSystemPathBuilderClz()));
jobWorkDirPath = new Path(jobWorkingDir);
Assert.assertTrue(fileSystem.exists(jobWorkDirPath));
} finally {
diff --git a/source-jdbc/src/main/java/org/apache/kylin/source/jdbc/JdbcHiveMRInput.java b/source-jdbc/src/main/java/org/apache/kylin/source/jdbc/JdbcHiveMRInput.java
index 11eb6f8..767e736 100644
--- a/source-jdbc/src/main/java/org/apache/kylin/source/jdbc/JdbcHiveMRInput.java
+++ b/source-jdbc/src/main/java/org/apache/kylin/source/jdbc/JdbcHiveMRInput.java
@@ -66,7 +66,7 @@ public class JdbcHiveMRInput extends HiveMRInput {
protected void addStepPhase1_DoCreateFlatTable(DefaultChainedExecutable jobFlow) {
final String cubeName = CubingExecutableUtil.getCubeName(jobFlow.getParams());
final String hiveInitStatements = JoinedFlatTable.generateHiveInitStatements(flatTableDatabase);
- final String jobWorkingDir = getJobWorkingDir(jobFlow, hdfsWorkingDir);
+ final String jobWorkingDir = getJobWorkingDir(jobFlow, hdfsWorkingDir, pathBuilder);
jobFlow.addTask(createSqoopToFlatHiveStep(jobWorkingDir, cubeName));
jobFlow.addTask(createFlatHiveTableFromFiles(hiveInitStatements, jobWorkingDir));
diff --git a/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java b/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java
index 1c94f9c..68f87ff 100644
--- a/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java
+++ b/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java
@@ -32,11 +32,11 @@ import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.util.Bytes;
+import org.apache.kylin.common.util.ClassUtil;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.cube.model.CubeJoinedFlatTableDesc;
import org.apache.kylin.engine.mr.IMRInput;
-import org.apache.kylin.engine.mr.JobBuilderSupport;
import org.apache.kylin.engine.mr.common.BatchConstants;
import org.apache.kylin.job.JoinedFlatTable;
import org.apache.kylin.job.engine.JobEngineConfig;
@@ -45,6 +45,7 @@ import org.apache.kylin.metadata.MetadataConstants;
import org.apache.kylin.metadata.model.IJoinedFlatTableDesc;
import org.apache.kylin.metadata.model.ISegment;
import org.apache.kylin.metadata.model.TableDesc;
+import org.apache.kylin.storage.path.IStoragePathBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -76,10 +77,12 @@ public class KafkaMRInput extends KafkaInputBase implements IMRInput {
private final CubeSegment cubeSegment;
private final JobEngineConfig conf;
private String delimiter = BatchConstants.SEQUENCE_FILE_DEFAULT_DELIMITER;
+ private final IStoragePathBuilder pathBuilder;
public KafkaTableInputFormat(CubeSegment cubeSegment, JobEngineConfig conf) {
this.cubeSegment = cubeSegment;
this.conf = conf;
+ this.pathBuilder = (IStoragePathBuilder)ClassUtil.newInstance(conf.getConfig().getStorageSystemPathBuilderClz());
}
@Override
@@ -88,7 +91,7 @@ public class KafkaMRInput extends KafkaInputBase implements IMRInput {
String jobId = job.getConfiguration().get(BatchConstants.ARG_CUBING_JOB_ID);
IJoinedFlatTableDesc flatHiveTableDesc = new CubeJoinedFlatTableDesc(cubeSegment);
String inputPath = JoinedFlatTable.getTableDir(flatHiveTableDesc,
- JobBuilderSupport.getJobWorkingDir(conf, jobId));
+ pathBuilder.getJobWorkingDir(conf.getHdfsWorkingDirectory(), jobId));
try {
FileInputFormat.addInputPath(job, new Path(inputPath));
} catch (IOException e) {
@@ -121,6 +124,7 @@ public class KafkaMRInput extends KafkaInputBase implements IMRInput {
private List<String> intermediateTables = Lists.newArrayList();
private List<String> intermediatePaths = Lists.newArrayList();
private String cubeName;
+ private IStoragePathBuilder pathBuilder;
public BatchCubingInputSide(CubeSegment seg, IJoinedFlatTableDesc flatDesc) {
this.conf = new JobEngineConfig(KylinConfig.getInstanceFromEnv());
@@ -130,6 +134,7 @@ public class KafkaMRInput extends KafkaInputBase implements IMRInput {
this.seg = seg;
this.cubeDesc = seg.getCubeDesc();
this.cubeName = seg.getCubeInstance().getName();
+ this.pathBuilder = (IStoragePathBuilder)ClassUtil.newInstance(config.getStorageSystemPathBuilderClz());
}
@Override
@@ -153,7 +158,7 @@ public class KafkaMRInput extends KafkaInputBase implements IMRInput {
}
protected String getJobWorkingDir(DefaultChainedExecutable jobFlow) {
- return JobBuilderSupport.getJobWorkingDir(config.getHdfsWorkingDirectory(), jobFlow.getId());
+ return pathBuilder.getJobWorkingDir(config.getHdfsWorkingDirectory(), jobFlow.getId());
}
@Override
diff --git a/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaSparkInput.java b/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaSparkInput.java
index 7db6c32..9c14fd6 100644
--- a/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaSparkInput.java
+++ b/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaSparkInput.java
@@ -21,15 +21,16 @@ import java.util.List;
import java.util.Locale;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.ClassUtil;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.model.CubeDesc;
-import org.apache.kylin.engine.mr.JobBuilderSupport;
import org.apache.kylin.engine.spark.ISparkInput;
import org.apache.kylin.job.engine.JobEngineConfig;
import org.apache.kylin.job.execution.DefaultChainedExecutable;
import org.apache.kylin.metadata.MetadataConstants;
import org.apache.kylin.metadata.model.IJoinedFlatTableDesc;
import org.apache.kylin.metadata.model.ISegment;
+import org.apache.kylin.storage.path.IStoragePathBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -62,6 +63,7 @@ public class KafkaSparkInput extends KafkaInputBase implements ISparkInput {
final private List<String> intermediateTables = Lists.newArrayList();
final private List<String> intermediatePaths = Lists.newArrayList();
private String cubeName;
+ private IStoragePathBuilder pathBuilder;
public BatchCubingInputSide(CubeSegment seg, IJoinedFlatTableDesc flatDesc) {
this.conf = new JobEngineConfig(KylinConfig.getInstanceFromEnv());
@@ -71,6 +73,7 @@ public class KafkaSparkInput extends KafkaInputBase implements ISparkInput {
this.seg = seg;
this.cubeDesc = seg.getCubeDesc();
this.cubeName = seg.getCubeInstance().getName();
+ this.pathBuilder = (IStoragePathBuilder)ClassUtil.newInstance(this.config.getStorageSystemPathBuilderClz());
}
@Override
@@ -94,7 +97,7 @@ public class KafkaSparkInput extends KafkaInputBase implements ISparkInput {
}
protected String getJobWorkingDir(DefaultChainedExecutable jobFlow) {
- return JobBuilderSupport.getJobWorkingDir(config.getHdfsWorkingDirectory(), jobFlow.getId());
+ return pathBuilder.getJobWorkingDir(config.getHdfsWorkingDirectory(), jobFlow.getId());
}
@Override
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/lookup/HBaseLookupMRSteps.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/lookup/HBaseLookupMRSteps.java
index 69e5e66..f6d11bd 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/lookup/HBaseLookupMRSteps.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/lookup/HBaseLookupMRSteps.java
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.Set;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.ClassUtil;
import org.apache.kylin.common.util.RandomUtil;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.model.CubeDesc;
@@ -45,6 +46,7 @@ import org.apache.kylin.metadata.model.TableRef;
import org.apache.kylin.source.IReadableTable;
import org.apache.kylin.source.SourceManager;
import org.apache.kylin.storage.hbase.HBaseConnection;
+import org.apache.kylin.storage.path.IStoragePathBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,10 +56,12 @@ public class HBaseLookupMRSteps {
protected static final Logger logger = LoggerFactory.getLogger(HBaseLookupMRSteps.class);
private CubeInstance cube;
private JobEngineConfig config;
+ private IStoragePathBuilder pathBuilder;
public HBaseLookupMRSteps(CubeInstance cube) {
this.cube = cube;
this.config = new JobEngineConfig(cube.getConfig());
+ this.pathBuilder = (IStoragePathBuilder)ClassUtil.newInstance(cube.getConfig().getStorageSystemPathBuilderClz());
}
public void addMaterializeLookupTablesSteps(LookupMaterializeContext context) {
@@ -158,7 +162,7 @@ public class HBaseLookupMRSteps {
}
private String getLookupTableHFilePath(String tableName, String jobId) {
- return HBaseConnection.makeQualifiedPathInHBaseCluster(JobBuilderSupport.getJobWorkingDir(config, jobId) + "/"
+ return HBaseConnection.makeQualifiedPathInHBaseCluster(pathBuilder.getJobWorkingDir(config.getHdfsWorkingDirectory(), jobId) + "/"
+ tableName + "/hfile/");
}
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HDFSPathGarbageCollectionStep.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HDFSPathGarbageCollectionStep.java
index 3ec27d4..05df80d 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HDFSPathGarbageCollectionStep.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HDFSPathGarbageCollectionStep.java
@@ -25,14 +25,15 @@ import java.util.List;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.kylin.common.util.ClassUtil;
import org.apache.kylin.common.util.HadoopUtil;
-import org.apache.kylin.engine.mr.JobBuilderSupport;
import org.apache.kylin.job.engine.JobEngineConfig;
import org.apache.kylin.job.exception.ExecuteException;
import org.apache.kylin.job.execution.AbstractExecutable;
import org.apache.kylin.job.execution.ExecutableContext;
import org.apache.kylin.job.execution.ExecuteResult;
import org.apache.kylin.storage.hbase.HBaseConnection;
+import org.apache.kylin.storage.path.IStoragePathBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -48,6 +49,7 @@ public class HDFSPathGarbageCollectionStep extends AbstractExecutable {
public static final String TO_DELETE_PATHS = "toDeletePaths";
private StringBuffer output;
private JobEngineConfig config;
+ private IStoragePathBuilder pathBuilder;
public HDFSPathGarbageCollectionStep() {
super();
@@ -58,6 +60,7 @@ public class HDFSPathGarbageCollectionStep extends AbstractExecutable {
protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException {
try {
config = new JobEngineConfig(context.getConfig());
+ pathBuilder = (IStoragePathBuilder)ClassUtil.newInstance(context.getConfig().getStorageSystemPathBuilderClz());
List<String> toDeletePaths = getDeletePaths();
dropHdfsPathOnCluster(toDeletePaths, HadoopUtil.getWorkingFileSystem());
@@ -93,7 +96,7 @@ public class HDFSPathGarbageCollectionStep extends AbstractExecutable {
// If hbase was deployed on another cluster, the job dir is empty and should be dropped,
// because of rowkey_stats and hfile dirs are both dropped.
if (fileSystem.listStatus(oldPath.getParent()).length == 0) {
- Path emptyJobPath = new Path(JobBuilderSupport.getJobWorkingDir(config, getJobId()));
+ Path emptyJobPath = new Path(pathBuilder.getJobWorkingDir(config.getHdfsWorkingDirectory(), getJobId()));
emptyJobPath = Path.getPathWithoutSchemeAndAuthority(emptyJobPath);
if (fileSystem.exists(emptyJobPath)) {
fileSystem.delete(emptyJobPath, true);
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/CubeMigrationCLI.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/CubeMigrationCLI.java
index 00635ba..ce43456 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/CubeMigrationCLI.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/CubeMigrationCLI.java
@@ -48,6 +48,7 @@ import org.apache.kylin.common.persistence.ResourceStore;
import org.apache.kylin.common.persistence.Serializer;
import org.apache.kylin.common.restclient.RestClient;
import org.apache.kylin.common.util.Bytes;
+import org.apache.kylin.common.util.ClassUtil;
import org.apache.kylin.common.util.Dictionary;
import org.apache.kylin.common.util.HadoopUtil;
import org.apache.kylin.cube.CubeInstance;
@@ -58,7 +59,6 @@ import org.apache.kylin.dict.DictionaryInfo;
import org.apache.kylin.dict.DictionaryManager;
import org.apache.kylin.dict.lookup.SnapshotManager;
import org.apache.kylin.dict.lookup.SnapshotTable;
-import org.apache.kylin.engine.mr.JobBuilderSupport;
import org.apache.kylin.metadata.cachesync.Broadcaster;
import org.apache.kylin.metadata.model.DataModelDesc;
import org.apache.kylin.metadata.model.SegmentStatusEnum;
@@ -68,6 +68,7 @@ import org.apache.kylin.metadata.realization.IRealizationConstants;
import org.apache.kylin.metadata.realization.RealizationStatusEnum;
import org.apache.kylin.metadata.realization.RealizationType;
import org.apache.kylin.storage.hbase.HBaseConnection;
+import org.apache.kylin.storage.path.IStoragePathBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -185,11 +186,13 @@ public class CubeMigrationCLI {
}
private static void renameFoldersInHdfs(CubeInstance cube) {
+ IStoragePathBuilder srcPathBuilder = (IStoragePathBuilder)ClassUtil.newInstance(srcConfig.getStorageSystemPathBuilderClz());
+ IStoragePathBuilder dstPathBuilder = (IStoragePathBuilder)ClassUtil.newInstance(dstConfig.getStorageSystemPathBuilderClz());
for (CubeSegment segment : cube.getSegments()) {
String jobUuid = segment.getLastBuildJobID();
- String src = JobBuilderSupport.getJobWorkingDir(srcConfig.getHdfsWorkingDirectory(), jobUuid);
- String tgt = JobBuilderSupport.getJobWorkingDir(dstConfig.getHdfsWorkingDirectory(), jobUuid);
+ String src = srcPathBuilder.getJobWorkingDir(srcConfig.getHdfsWorkingDirectory(), jobUuid);
+ String tgt = dstPathBuilder.getJobWorkingDir(dstConfig.getHdfsWorkingDirectory(), jobUuid);
operations.add(new Opt(OptType.RENAME_FOLDER_IN_HDFS, new Object[] { src, tgt }));
}
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/StorageCleanupJob.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/StorageCleanupJob.java
index 725d819..fd6124b 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/StorageCleanupJob.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/StorageCleanupJob.java
@@ -45,6 +45,7 @@ import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.util.AbstractApplication;
+import org.apache.kylin.common.util.ClassUtil;
import org.apache.kylin.common.util.CliCommandExecutor;
import org.apache.kylin.common.util.HadoopUtil;
import org.apache.kylin.common.util.HiveCmdBuilder;
@@ -53,7 +54,6 @@ import org.apache.kylin.common.util.Pair;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.engine.mr.JobBuilderSupport;
import org.apache.kylin.job.engine.JobEngineConfig;
import org.apache.kylin.job.execution.AbstractExecutable;
import org.apache.kylin.job.execution.ExecutableManager;
@@ -61,6 +61,7 @@ import org.apache.kylin.job.execution.ExecutableState;
import org.apache.kylin.metadata.MetadataConstants;
import org.apache.kylin.metadata.realization.IRealizationConstants;
import org.apache.kylin.storage.hbase.HBaseConnection;
+import org.apache.kylin.storage.path.IStoragePathBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -186,6 +187,7 @@ public class StorageCleanupJob extends AbstractApplication {
private void cleanUnusedHdfsFiles(Configuration conf) throws IOException {
JobEngineConfig engineConfig = new JobEngineConfig(KylinConfig.getInstanceFromEnv());
CubeManager cubeMgr = CubeManager.getInstance(KylinConfig.getInstanceFromEnv());
+ IStoragePathBuilder pathBuilder = (IStoragePathBuilder)ClassUtil.newInstance(engineConfig.getConfig().getStorageSystemPathBuilderClz());
FileSystem fs = HadoopUtil.getWorkingFileSystem(conf);
List<String> allHdfsPathsNeedToBeDeleted = new ArrayList<String>();
@@ -208,7 +210,7 @@ public class StorageCleanupJob extends AbstractApplication {
// only remove FINISHED and DISCARDED job intermediate files
final ExecutableState state = executableManager.getOutput(jobId).getState();
if (!state.isFinalState()) {
- String path = JobBuilderSupport.getJobWorkingDir(engineConfig.getHdfsWorkingDirectory(), jobId);
+ String path = pathBuilder.getJobWorkingDir(engineConfig.getHdfsWorkingDirectory(), jobId);
allHdfsPathsNeedToBeDeleted.remove(path);
logger.info("Skip " + path + " from deletion list, as the path belongs to job " + jobId + " with status " + state);
}
@@ -219,7 +221,7 @@ public class StorageCleanupJob extends AbstractApplication {
for (CubeSegment seg : cube.getSegments()) {
String jobUuid = seg.getLastBuildJobID();
if (jobUuid != null && jobUuid.equals("") == false) {
- String path = JobBuilderSupport.getJobWorkingDir(engineConfig.getHdfsWorkingDirectory(), jobUuid);
+ String path = pathBuilder.getJobWorkingDir(engineConfig.getHdfsWorkingDirectory(), jobUuid);
allHdfsPathsNeedToBeDeleted.remove(path);
logger.info("Skip " + path + " from deletion list, as the path belongs to segment " + seg + " of cube " + cube.getName());
}
diff --git a/storage-parquet/src/main/java/org/apache/kylin/storage/parquet/steps/ParquetMRSteps.java b/storage-parquet/src/main/java/org/apache/kylin/storage/parquet/steps/ParquetMRSteps.java
index 829f074..e1d0b0f 100644
--- a/storage-parquet/src/main/java/org/apache/kylin/storage/parquet/steps/ParquetMRSteps.java
+++ b/storage-parquet/src/main/java/org/apache/kylin/storage/parquet/steps/ParquetMRSteps.java
@@ -45,7 +45,7 @@ public class ParquetMRSteps extends ParquetJobSteps{
appendMapReduceParameters(cmd);
appendExecCmdParameters(cmd, BatchConstants.ARG_CUBE_NAME, seg.getRealization().getName());
appendExecCmdParameters(cmd, BatchConstants.ARG_INPUT, inputPath);
- appendExecCmdParameters(cmd, BatchConstants.ARG_OUTPUT, getParquetOutputPath(jobId));
+ appendExecCmdParameters(cmd, BatchConstants.ARG_OUTPUT, getParquetOutputPath());
appendExecCmdParameters(cmd, BatchConstants.ARG_SEGMENT_ID, seg.getUuid());
appendExecCmdParameters(cmd, BatchConstants.ARG_JOB_NAME, "Kylin_Parquet_Generator_" + seg.getRealization().getName()+ "_Step");
diff --git a/storage-parquet/src/main/java/org/apache/kylin/storage/parquet/steps/ParquetSparkSteps.java b/storage-parquet/src/main/java/org/apache/kylin/storage/parquet/steps/ParquetSparkSteps.java
index 65bd30a..6486187 100644
--- a/storage-parquet/src/main/java/org/apache/kylin/storage/parquet/steps/ParquetSparkSteps.java
+++ b/storage-parquet/src/main/java/org/apache/kylin/storage/parquet/steps/ParquetSparkSteps.java
@@ -46,7 +46,7 @@ public class ParquetSparkSteps extends ParquetJobSteps {
sparkExecutable.setParam(SparkCubeParquet.OPTION_CUBE_NAME.getOpt(), seg.getRealization().getName());
sparkExecutable.setParam(SparkCubeParquet.OPTION_SEGMENT_ID.getOpt(), seg.getUuid());
sparkExecutable.setParam(SparkCubeParquet.OPTION_INPUT_PATH.getOpt(), inputPath);
- sparkExecutable.setParam(SparkCubeParquet.OPTION_OUTPUT_PATH.getOpt(), getParquetOutputPath(jobId));
+ sparkExecutable.setParam(SparkCubeParquet.OPTION_OUTPUT_PATH.getOpt(), getParquetOutputPath());
sparkExecutable.setParam(SparkCubeParquet.OPTION_META_URL.getOpt(),
jobBuilder2.getSegmentMetadataUrl(seg.getConfig(), jobId));
sparkExecutable.setJobId(jobId);
diff --git a/tool/src/main/java/org/apache/kylin/tool/CubeMigrationCLI.java b/tool/src/main/java/org/apache/kylin/tool/CubeMigrationCLI.java
index 6909b74..5c2a839 100644
--- a/tool/src/main/java/org/apache/kylin/tool/CubeMigrationCLI.java
+++ b/tool/src/main/java/org/apache/kylin/tool/CubeMigrationCLI.java
@@ -42,6 +42,7 @@ import org.apache.kylin.common.persistence.ResourceStore;
import org.apache.kylin.common.persistence.Serializer;
import org.apache.kylin.common.restclient.RestClient;
import org.apache.kylin.common.util.AbstractApplication;
+import org.apache.kylin.common.util.ClassUtil;
import org.apache.kylin.common.util.Dictionary;
import org.apache.kylin.common.util.HadoopUtil;
import org.apache.kylin.common.util.OptionsHelper;
@@ -53,7 +54,6 @@ import org.apache.kylin.dict.DictionaryInfo;
import org.apache.kylin.dict.DictionaryManager;
import org.apache.kylin.dict.lookup.SnapshotManager;
import org.apache.kylin.dict.lookup.SnapshotTable;
-import org.apache.kylin.engine.mr.JobBuilderSupport;
import org.apache.kylin.metadata.MetadataConstants;
import org.apache.kylin.metadata.model.DataModelDesc;
import org.apache.kylin.metadata.model.IStorageAware;
@@ -66,6 +66,7 @@ import org.apache.kylin.metadata.realization.IRealizationConstants;
import org.apache.kylin.metadata.realization.RealizationStatusEnum;
import org.apache.kylin.metadata.realization.RealizationType;
import org.apache.kylin.storage.hbase.HBaseConnection;
+import org.apache.kylin.storage.path.IStoragePathBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -229,11 +230,14 @@ public class CubeMigrationCLI extends AbstractApplication {
}
protected void renameFoldersInHdfs(CubeInstance cube) throws IOException {
+ IStoragePathBuilder srcPathBuilder = (IStoragePathBuilder)ClassUtil.newInstance(srcConfig.getStorageSystemPathBuilderClz());
+ IStoragePathBuilder dstPathBuilder = (IStoragePathBuilder)ClassUtil.newInstance(dstConfig.getStorageSystemPathBuilderClz());
+
for (CubeSegment segment : cube.getSegments()) {
String jobUuid = segment.getLastBuildJobID();
- String src = JobBuilderSupport.getJobWorkingDir(srcConfig.getHdfsWorkingDirectory(), jobUuid);
- String tgt = JobBuilderSupport.getJobWorkingDir(dstConfig.getHdfsWorkingDirectory(), jobUuid);
+ String src = srcPathBuilder.getJobWorkingDir(srcConfig.getHdfsWorkingDirectory(), jobUuid);
+ String tgt = dstPathBuilder.getJobWorkingDir(dstConfig.getHdfsWorkingDirectory(), jobUuid);
operations.add(new Opt(OptType.RENAME_FOLDER_IN_HDFS, new Object[] { src, tgt }));
}