kylin-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From liy...@apache.org
Subject [1/5] kylin git commit: KYLIN-2351 Support using non-default HDFS path as kylin.env.hdfs-working-dir [Forced Update!]
Date Wed, 04 Jan 2017 14:34:45 GMT
Repository: kylin
Updated Branches:
  refs/heads/yang22-cdh5.7 ee32cb13b -> e9a6dcc5a (forced update)


KYLIN-2351 Support using non-default HDFS path as kylin.env.hdfs-working-dir


Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/fa6cbe5b
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/fa6cbe5b
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/fa6cbe5b

Branch: refs/heads/yang22-cdh5.7
Commit: fa6cbe5bf1d96e48f6cf88b020b954a63a2845c5
Parents: ee04dca
Author: Li Yang <liyang@apache.org>
Authored: Wed Jan 4 18:24:48 2017 +0800
Committer: Li Yang <liyang@apache.org>
Committed: Wed Jan 4 18:24:48 2017 +0800

----------------------------------------------------------------------
 .../org/apache/kylin/common/KylinConfigBase.java     |  9 +++++++++
 .../java/org/apache/kylin/engine/mr/HadoopUtil.java  | 15 ++++++++++++++-
 .../kylin/engine/mr/common/AbstractHadoopJob.java    |  4 ++--
 .../kylin/engine/mr/steps/CreateDictionaryJob.java   |  2 +-
 .../engine/mr/steps/FactDistinctColumnsReducer.java  | 10 ++++++----
 .../kylin/engine/mr/steps/SaveStatisticsStep.java    |  2 +-
 .../apache/kylin/provision/BuildCubeWithEngine.java  |  3 +--
 .../org/apache/kylin/query/ITMassInQueryTest.java    |  3 +--
 .../apache/kylin/storage/hbase/ITHdfsOpsTest.java    |  4 +---
 .../org/apache/kylin/source/hive/HiveMRInput.java    |  2 +-
 .../org/apache/kylin/source/kafka/KafkaMRInput.java  |  2 +-
 .../kylin/storage/hbase/HBaseResourceStore.java      | 11 ++++-------
 .../kylin/storage/hbase/steps/DeprecatedGCStep.java  |  3 +--
 .../hbase/steps/HDFSPathGarbageCollectionStep.java   |  2 +-
 .../hbase/steps/SequenceFileCuboidWriter.java        |  2 +-
 .../kylin/storage/hbase/util/CubeMigrationCLI.java   |  3 ++-
 .../kylin/storage/hbase/util/StorageCleanupJob.java  |  3 ++-
 .../java/org/apache/kylin/tool/CubeMigrationCLI.java |  3 ++-
 .../org/apache/kylin/tool/StorageCleanupJob.java     |  4 ++--
 19 files changed, 53 insertions(+), 34 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
----------------------------------------------------------------------
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 194f158..5b7c734 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
@@ -171,6 +171,15 @@ abstract public class KylinConfigBase implements Serializable {
         if (!root.endsWith("/")) {
             root += "/";
         }
+        
+        // make sure path qualified
+        if (!root.contains("://")) {
+            if (!root.startsWith("/"))
+                root = "hdfs:///" + root;
+            else
+                root = "hdfs://" + root;
+        }
+        
         return new StringBuffer(root).append(StringUtils.replaceChars(getMetadataUrlPrefix(),
':', '-')).append("/").toString();
     }
 

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/engine-mr/src/main/java/org/apache/kylin/engine/mr/HadoopUtil.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/HadoopUtil.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/HadoopUtil.java
index 88692a0..3d29a02 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/HadoopUtil.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/HadoopUtil.java
@@ -30,6 +30,7 @@ import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.io.Writable;
+import org.apache.kylin.common.KylinConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -66,10 +67,22 @@ public class HadoopUtil {
         return conf;
     }
 
+    public static FileSystem getWorkingFileSystem() throws IOException {
+        return getFileSystem(KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory());
+    }
+    
+    public static FileSystem getWorkingFileSystem(Configuration conf) throws IOException
{
+        return getFileSystem(KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory(),
conf);
+    }
+    
     public static FileSystem getFileSystem(String path) throws IOException {
-        return FileSystem.get(makeURI(path), getCurrentConfiguration());
+        return getFileSystem(path, getCurrentConfiguration());
     }
 
+    static FileSystem getFileSystem(String path, Configuration conf) throws IOException {
+        return FileSystem.get(makeURI(path), conf);
+    }
+    
     public static URI makeURI(String filePath) {
         try {
             return new URI(fixWindowsPath(filePath));

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/AbstractHadoopJob.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/AbstractHadoopJob.java
b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/AbstractHadoopJob.java
index 80636d3..243e59d 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/AbstractHadoopJob.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/AbstractHadoopJob.java
@@ -284,7 +284,7 @@ public abstract class AbstractHadoopJob extends Configured implements
Tool {
         try {
             Configuration jobConf = job.getConfiguration();
             FileSystem localfs = FileSystem.getLocal(jobConf);
-            FileSystem hdfs = FileSystem.get(jobConf);
+            FileSystem hdfs = HadoopUtil.getWorkingFileSystem(jobConf);
 
             StringBuilder jarList = new StringBuilder();
             StringBuilder fileList = new StringBuilder();
@@ -408,7 +408,7 @@ public abstract class AbstractHadoopJob extends Configured implements
Tool {
             inp = inp.trim();
             if (inp.endsWith("/*")) {
                 inp = inp.substring(0, inp.length() - 2);
-                FileSystem fs = FileSystem.get(job.getConfiguration());
+                FileSystem fs = HadoopUtil.getWorkingFileSystem(job.getConfiguration());
                 Path path = new Path(inp);
 
                 if (!exists(fs, path)) {

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CreateDictionaryJob.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CreateDictionaryJob.java
b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CreateDictionaryJob.java
index 4985503..8187ff8 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CreateDictionaryJob.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CreateDictionaryJob.java
@@ -65,7 +65,7 @@ public class CreateDictionaryJob extends AbstractHadoopJob {
             public Dictionary<String> getDictionary(TblColRef col) throws IOException
{
                 Path colDir = new Path(factColumnsInputPath, col.getName());
                 Path dictFile = new Path(colDir, col.getName() + FactDistinctColumnsReducer.DICT_FILE_POSTFIX);
-                FileSystem fs = HadoopUtil.getFileSystem(dictFile.toString());
+                FileSystem fs = HadoopUtil.getWorkingFileSystem();
                 if (fs.exists(dictFile) == false)
                     return null;
                 

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsReducer.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsReducer.java
b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsReducer.java
index a59e3ae..6ef976e 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsReducer.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctColumnsReducer.java
@@ -43,6 +43,7 @@ import org.apache.kylin.cube.CubeManager;
 import org.apache.kylin.cube.model.CubeDesc;
 import org.apache.kylin.dict.DictionaryGenerator;
 import org.apache.kylin.dict.IDictionaryBuilder;
+import org.apache.kylin.engine.mr.HadoopUtil;
 import org.apache.kylin.engine.mr.KylinReducer;
 import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
 import org.apache.kylin.engine.mr.common.BatchConstants;
@@ -199,7 +200,7 @@ public class FactDistinctColumnsReducer extends KylinReducer<SelfDefineSortableK
 
     private void outputDistinctValues(TblColRef col, Collection<ByteArray> values,
Context context) throws IOException {
         final Configuration conf = context.getConfiguration();
-        final FileSystem fs = FileSystem.get(conf);
+        final FileSystem fs = HadoopUtil.getWorkingFileSystem(conf);
         final String outputPath = conf.get(BatchConstants.CFG_OUTPUT_PATH);
         final Path colDir = new Path(outputPath, col.getName());
         final String fileName = col.getName() + "-" + taskId % uhcReducerCount;
@@ -255,7 +256,7 @@ public class FactDistinctColumnsReducer extends KylinReducer<SelfDefineSortableK
 
     private FSDataOutputStream getOutputStream(Context context, String outputFileName) throws
IOException {
         final Configuration conf = context.getConfiguration();
-        final FileSystem fs = FileSystem.get(conf);
+        final FileSystem fs = HadoopUtil.getWorkingFileSystem(conf);
         final String outputPath = conf.get(BatchConstants.CFG_OUTPUT_PATH);
         final Path colDir = new Path(outputPath, col.getName());
         final Path outputFile = new Path(colDir, outputFileName);
@@ -304,8 +305,9 @@ public class FactDistinctColumnsReducer extends KylinReducer<SelfDefineSortableK
 
     private void writeMapperAndCuboidStatistics(Context context) throws IOException {
         Configuration conf = context.getConfiguration();
-        FileSystem fs = FileSystem.get(conf);
-        FSDataOutputStream out = fs.create(new Path(statisticsOutput, BatchConstants.CFG_STATISTICS_CUBE_ESTIMATION_FILENAME));
+        FileSystem fs = HadoopUtil.getWorkingFileSystem(conf);
+        Path path = new Path(statisticsOutput, BatchConstants.CFG_STATISTICS_CUBE_ESTIMATION_FILENAME);
+        FSDataOutputStream out = fs.create(path);
 
         try {
             String msg;

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/SaveStatisticsStep.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/SaveStatisticsStep.java
b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/SaveStatisticsStep.java
index 020c62c..6120270 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/SaveStatisticsStep.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/SaveStatisticsStep.java
@@ -59,8 +59,8 @@ public class SaveStatisticsStep extends AbstractExecutable {
 
         ResourceStore rs = ResourceStore.getStore(kylinConf);
         try {
+            FileSystem fs = HadoopUtil.getWorkingFileSystem();
             Path statisticsFilePath = new Path(CubingExecutableUtil.getStatisticsPath(this.getParams()),
BatchConstants.CFG_STATISTICS_CUBOID_ESTIMATION_FILENAME);
-            FileSystem fs = FileSystem.get(HadoopUtil.getCurrentConfiguration());
             if (!fs.exists(statisticsFilePath))
                 throw new IOException("File " + statisticsFilePath + " does not exists");
 

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithEngine.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithEngine.java b/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithEngine.java
index 67b62d5..7a2954d 100644
--- a/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithEngine.java
+++ b/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithEngine.java
@@ -114,8 +114,7 @@ public class BuildCubeWithEngine {
 
         try {
             //check hdfs permission
-            Configuration hconf = HadoopUtil.getCurrentConfiguration();
-            FileSystem fileSystem = FileSystem.get(hconf);
+            FileSystem fileSystem = HadoopUtil.getWorkingFileSystem();
             String hdfsWorkingDirectory = KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory();
             Path coprocessorDir = new Path(hdfsWorkingDirectory);
             boolean success = fileSystem.mkdirs(coprocessorDir);

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/kylin-it/src/test/java/org/apache/kylin/query/ITMassInQueryTest.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/query/ITMassInQueryTest.java b/kylin-it/src/test/java/org/apache/kylin/query/ITMassInQueryTest.java
index 5835baf..18d79ae 100644
--- a/kylin-it/src/test/java/org/apache/kylin/query/ITMassInQueryTest.java
+++ b/kylin-it/src/test/java/org/apache/kylin/query/ITMassInQueryTest.java
@@ -68,8 +68,7 @@ public class ITMassInQueryTest extends KylinTestBase {
         ITKylinQueryTest.joinType = "left";
         ITKylinQueryTest.setupAll();
 
-        Configuration hconf = HadoopUtil.getCurrentConfiguration();
-        fileSystem = FileSystem.get(hconf);
+        fileSystem = HadoopUtil.getWorkingFileSystem();
 
         int sellerCount = 200;
         Random r = new Random();

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITHdfsOpsTest.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITHdfsOpsTest.java b/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITHdfsOpsTest.java
index 499a456..786d7d1 100644
--- a/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITHdfsOpsTest.java
+++ b/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITHdfsOpsTest.java
@@ -42,9 +42,7 @@ public class ITHdfsOpsTest extends HBaseMetadataTestCase {
 
         this.createTestMetadata();
 
-        Configuration hconf = HadoopUtil.getCurrentConfiguration();
-
-        fileSystem = FileSystem.get(hconf);
+        fileSystem = HadoopUtil.getWorkingFileSystem();
     }
 
     @Test

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java
----------------------------------------------------------------------
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 fc2b982..0f351b0 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
@@ -379,7 +379,7 @@ public class HiveMRInput implements IMRInput {
 
         private void rmdirOnHDFS(String path) throws IOException {
             Path externalDataPath = new Path(path);
-            FileSystem fs = FileSystem.get(externalDataPath.toUri(), HadoopUtil.getCurrentConfiguration());
+            FileSystem fs = HadoopUtil.getWorkingFileSystem();
             if (fs.exists(externalDataPath)) {
                 fs.delete(externalDataPath, true);
             }

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/source-kafka/src/main/java/org/apache/kylin/source/kafka/KafkaMRInput.java
----------------------------------------------------------------------
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 dde3584..2f4fded 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
@@ -223,7 +223,7 @@ public class KafkaMRInput implements IMRInput {
 
         private void rmdirOnHDFS(String path) throws IOException {
             Path externalDataPath = new Path(path);
-            FileSystem fs = FileSystem.get(externalDataPath.toUri(), HadoopUtil.getCurrentConfiguration());
+            FileSystem fs = HadoopUtil.getWorkingFileSystem();
             if (fs.exists(externalDataPath)) {
                 fs.delete(externalDataPath, true);
             }

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseResourceStore.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseResourceStore.java
b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseResourceStore.java
index 1d19983..2a12984 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseResourceStore.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseResourceStore.java
@@ -27,7 +27,6 @@ import java.util.NavigableSet;
 import java.util.TreeSet;
 
 import org.apache.commons.io.IOUtils;
-import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -49,6 +48,7 @@ import org.apache.kylin.common.persistence.RawResource;
 import org.apache.kylin.common.persistence.ResourceStore;
 import org.apache.kylin.common.util.Bytes;
 import org.apache.kylin.common.util.BytesUtil;
+import org.apache.kylin.engine.mr.HadoopUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -201,8 +201,7 @@ public class HBaseResourceStore extends ResourceStore {
         byte[] value = r.getValue(B_FAMILY, B_COLUMN);
         if (value.length == 0) {
             Path redirectPath = bigCellHDFSPath(resPath);
-            Configuration hconf = HBaseConnection.getCurrentHBaseConfiguration();
-            FileSystem fileSystem = FileSystem.get(hconf);
+            FileSystem fileSystem = HadoopUtil.getWorkingFileSystem();
 
             return fileSystem.open(redirectPath);
         } else {
@@ -292,8 +291,7 @@ public class HBaseResourceStore extends ResourceStore {
 
             if (hdfsResourceExist) { // remove hdfs cell value
                 Path redirectPath = bigCellHDFSPath(resPath);
-                Configuration hconf = HBaseConnection.getCurrentHBaseConfiguration();
-                FileSystem fileSystem = FileSystem.get(hconf);
+                FileSystem fileSystem = HadoopUtil.getWorkingFileSystem();
 
                 if (fileSystem.exists(redirectPath)) {
                     fileSystem.delete(redirectPath, true);
@@ -339,8 +337,7 @@ public class HBaseResourceStore extends ResourceStore {
 
     private Path writeLargeCellToHdfs(String resPath, byte[] largeColumn, HTableInterface
table) throws IOException {
         Path redirectPath = bigCellHDFSPath(resPath);
-        Configuration hconf = HBaseConnection.getCurrentHBaseConfiguration();
-        FileSystem fileSystem = FileSystem.get(hconf);
+        FileSystem fileSystem = HadoopUtil.getWorkingFileSystem();
 
         if (fileSystem.exists(redirectPath)) {
             fileSystem.delete(redirectPath, true);

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/DeprecatedGCStep.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/DeprecatedGCStep.java
b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/DeprecatedGCStep.java
index 46a828e..fbe64d9 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/DeprecatedGCStep.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/DeprecatedGCStep.java
@@ -132,8 +132,7 @@ public class DeprecatedGCStep extends AbstractExecutable {
 
         List<String> oldHdfsPaths = this.getOldHdsfPaths();
         if (oldHdfsPaths != null && oldHdfsPaths.size() > 0) {
-            Configuration hconf = HadoopUtil.getCurrentConfiguration();
-            FileSystem fileSystem = FileSystem.get(hconf);
+            FileSystem fileSystem = HadoopUtil.getWorkingFileSystem();
             for (String path : oldHdfsPaths) {
                 if (path.endsWith("*"))
                     path = path.substring(0, path.length() - 1);

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/HDFSPathGarbageCollectionStep.java
----------------------------------------------------------------------
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 fbfd582..89baf95 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
@@ -59,7 +59,7 @@ public class HDFSPathGarbageCollectionStep extends AbstractExecutable {
         try {
             config = new JobEngineConfig(context.getConfig());
             List<String> toDeletePaths = getDeletePaths();
-            dropHdfsPathOnCluster(toDeletePaths, FileSystem.get(HadoopUtil.getCurrentConfiguration()));
+            dropHdfsPathOnCluster(toDeletePaths, HadoopUtil.getWorkingFileSystem());
 
             if (StringUtils.isNotEmpty(context.getConfig().getHBaseClusterFs())) {
                 dropHdfsPathOnCluster(toDeletePaths, FileSystem.get(HBaseConnection.getCurrentHBaseConfiguration()));

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/SequenceFileCuboidWriter.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/SequenceFileCuboidWriter.java
b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/SequenceFileCuboidWriter.java
index 8f2fc80..1d66d3e 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/SequenceFileCuboidWriter.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/SequenceFileCuboidWriter.java
@@ -55,7 +55,7 @@ public class SequenceFileCuboidWriter extends KVGTRecordWriter {
             JobBuilderSupport jobBuilderSupport = new JobBuilderSupport(cubeSegment, "SYSTEM");
             String cuboidRoot = jobBuilderSupport.getCuboidRootPath(cubeSegment);
             Path cuboidPath = new Path(cuboidRoot);
-            FileSystem fs = HadoopUtil.getFileSystem(cuboidRoot);
+            FileSystem fs = HadoopUtil.getWorkingFileSystem();
             try {
                 if (fs.exists(cuboidPath)) {
                     fs.delete(cuboidPath, true);

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/CubeMigrationCLI.java
----------------------------------------------------------------------
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 2e682b1..31864f6 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
@@ -55,6 +55,7 @@ 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.HadoopUtil;
 import org.apache.kylin.engine.mr.JobBuilderSupport;
 import org.apache.kylin.metadata.cachesync.Broadcaster;
 import org.apache.kylin.metadata.model.DataModelDesc;
@@ -136,7 +137,7 @@ public class CubeMigrationCLI {
         Configuration conf = HBaseConnection.getCurrentHBaseConfiguration();
         hbaseAdmin = new HBaseAdmin(conf);
 
-        hdfsFS = FileSystem.get(new Configuration());
+        hdfsFS = HadoopUtil.getWorkingFileSystem();
 
         operations = new ArrayList<Opt>();
 

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/StorageCleanupJob.java
----------------------------------------------------------------------
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 9fe5a23..d1a74ad 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
@@ -50,6 +50,7 @@ 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.HadoopUtil;
 import org.apache.kylin.engine.mr.JobBuilderSupport;
 import org.apache.kylin.job.engine.JobEngineConfig;
 import org.apache.kylin.job.execution.AbstractExecutable;
@@ -180,7 +181,7 @@ public class StorageCleanupJob extends AbstractApplication {
         JobEngineConfig engineConfig = new JobEngineConfig(KylinConfig.getInstanceFromEnv());
         CubeManager cubeMgr = CubeManager.getInstance(KylinConfig.getInstanceFromEnv());
 
-        FileSystem fs = FileSystem.get(conf);
+        FileSystem fs = HadoopUtil.getWorkingFileSystem(conf);
         List<String> allHdfsPathsNeedToBeDeleted = new ArrayList<String>();
         // GlobFilter filter = new
         // GlobFilter(KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory()

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/tool/src/main/java/org/apache/kylin/tool/CubeMigrationCLI.java
----------------------------------------------------------------------
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 40306c9..5269195 100644
--- a/tool/src/main/java/org/apache/kylin/tool/CubeMigrationCLI.java
+++ b/tool/src/main/java/org/apache/kylin/tool/CubeMigrationCLI.java
@@ -55,6 +55,7 @@ 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.HadoopUtil;
 import org.apache.kylin.engine.mr.JobBuilderSupport;
 import org.apache.kylin.metadata.cachesync.Broadcaster;
 import org.apache.kylin.metadata.model.DataModelDesc;
@@ -134,7 +135,7 @@ public class CubeMigrationCLI {
         Configuration conf = HBaseConnection.getCurrentHBaseConfiguration();
         hbaseAdmin = new HBaseAdmin(conf);
 
-        hdfsFS = FileSystem.get(new Configuration());
+        hdfsFS = HadoopUtil.getWorkingFileSystem();
 
         operations = new ArrayList<Opt>();
 

http://git-wip-us.apache.org/repos/asf/kylin/blob/fa6cbe5b/tool/src/main/java/org/apache/kylin/tool/StorageCleanupJob.java
----------------------------------------------------------------------
diff --git a/tool/src/main/java/org/apache/kylin/tool/StorageCleanupJob.java b/tool/src/main/java/org/apache/kylin/tool/StorageCleanupJob.java
index 05e0142..b3e2ec0 100644
--- a/tool/src/main/java/org/apache/kylin/tool/StorageCleanupJob.java
+++ b/tool/src/main/java/org/apache/kylin/tool/StorageCleanupJob.java
@@ -185,7 +185,7 @@ public class StorageCleanupJob extends AbstractApplication {
         JobEngineConfig engineConfig = new JobEngineConfig(KylinConfig.getInstanceFromEnv());
         CubeManager cubeMgr = CubeManager.getInstance(KylinConfig.getInstanceFromEnv());
 
-        FileSystem fs = FileSystem.get(conf);
+        FileSystem fs = HadoopUtil.getWorkingFileSystem(conf);
         List<String> allHdfsPathsNeedToBeDeleted = new ArrayList<String>();
         // GlobFilter filter = new
         // GlobFilter(KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory()
@@ -342,7 +342,7 @@ public class StorageCleanupJob extends AbstractApplication {
                     if (segmentId2JobId.containsKey(segmentId)) {
                         String path = JobBuilderSupport.getJobWorkingDir(engineConfig.getHdfsWorkingDirectory(),
segmentId2JobId.get(segmentId)) + "/" + tableToDelete;
                         Path externalDataPath = new Path(path);
-                        FileSystem fs = FileSystem.get(externalDataPath.toUri(), HadoopUtil.getCurrentConfiguration());
+                        FileSystem fs = HadoopUtil.getWorkingFileSystem();
                         if (fs.exists(externalDataPath)) {
                             fs.delete(externalDataPath, true);
                             logger.info("Hive table {}'s external path {} deleted", tableToDelete,
path);


Mime
View raw message