kylin-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From liy...@apache.org
Subject [31/49] kylin git commit: KYLIN-2145 StorageCleanupJob will fail when beeline enabled
Date Sun, 06 Nov 2016 09:31:07 GMT
KYLIN-2145 StorageCleanupJob will fail when beeline enabled


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

Branch: refs/heads/KYLIN-1971
Commit: cad3def9e21fcfc07767e3a1b72f5b1bf8cae3de
Parents: 7358a78
Author: Hongbin Ma <mahongbin@apache.org>
Authored: Tue Nov 1 18:32:31 2016 +0800
Committer: Hongbin Ma <mahongbin@apache.org>
Committed: Thu Nov 3 13:34:12 2016 +0800

----------------------------------------------------------------------
 tool/pom.xml                                    |  4 ++
 .../apache/kylin/tool/StorageCleanupJob.java    | 44 ++++++++++----------
 2 files changed, 25 insertions(+), 23 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kylin/blob/cad3def9/tool/pom.xml
----------------------------------------------------------------------
diff --git a/tool/pom.xml b/tool/pom.xml
index e530469..e3d7bfa 100644
--- a/tool/pom.xml
+++ b/tool/pom.xml
@@ -42,6 +42,10 @@
             <groupId>org.apache.kylin</groupId>
             <artifactId>kylin-source-kafka</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.kylin</groupId>
+            <artifactId>kylin-source-hive</artifactId>
+        </dependency>
 
         <!--Env-->
         <dependency>

http://git-wip-us.apache.org/repos/asf/kylin/blob/cad3def9/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 c1ff753..56681af 100644
--- a/tool/src/main/java/org/apache/kylin/tool/StorageCleanupJob.java
+++ b/tool/src/main/java/org/apache/kylin/tool/StorageCleanupJob.java
@@ -18,9 +18,7 @@
 
 package org.apache.kylin.tool;
 
-import java.io.BufferedReader;
 import java.io.IOException;
-import java.io.StringReader;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.Callable;
@@ -31,6 +29,8 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.regex.Pattern;
 
+import javax.annotation.Nullable;
+
 import org.apache.commons.cli.Option;
 import org.apache.commons.cli.OptionBuilder;
 import org.apache.commons.cli.Options;
@@ -46,7 +46,6 @@ import org.apache.kylin.common.KylinConfig;
 import org.apache.kylin.common.util.AbstractApplication;
 import org.apache.kylin.common.util.CliCommandExecutor;
 import org.apache.kylin.common.util.OptionsHelper;
-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;
@@ -56,10 +55,15 @@ import org.apache.kylin.job.execution.AbstractExecutable;
 import org.apache.kylin.job.execution.ExecutableState;
 import org.apache.kylin.job.manager.ExecutableManager;
 import org.apache.kylin.metadata.realization.IRealizationConstants;
-import org.apache.kylin.storage.hbase.util.HiveCmdBuilder;
+import org.apache.kylin.source.hive.HiveClientFactory;
+import org.apache.kylin.source.hive.HiveCmdBuilder;
+import org.apache.kylin.source.hive.IHiveClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.base.Predicate;
+import com.google.common.collect.Iterables;
+
 public class StorageCleanupJob extends AbstractApplication {
 
     @SuppressWarnings("static-access")
@@ -244,27 +248,24 @@ public class StorageCleanupJob extends AbstractApplication {
             }
             System.out.println("-------------------------------------------------------");
         }
-
     }
 
-    private void cleanUnusedIntermediateHiveTable(Configuration conf) throws IOException
{
+    private void cleanUnusedIntermediateHiveTable(Configuration conf) throws Exception {
         final KylinConfig config = KylinConfig.getInstanceFromEnv();
         final CliCommandExecutor cmdExec = config.getCliCommandExecutor();
         final int uuidLength = 36;
         final String preFix = "kylin_intermediate_";
         final String uuidPattern = "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}";
 
-        
-        final String useDatabaseHql = "USE " + config.getHiveDatabaseForIntermediateTable()
+ ";";
-        final HiveCmdBuilder hiveCmdBuilder = new HiveCmdBuilder();
-        hiveCmdBuilder.addStatement(useDatabaseHql);
-        hiveCmdBuilder.addStatement("show tables " + "\'kylin_intermediate_*\'" + "; ");
-
-        Pair<Integer, String> result = cmdExec.execute(hiveCmdBuilder.build());
+        IHiveClient hiveClient = HiveClientFactory.getHiveClient();
+        List<String> hiveTableNames = hiveClient.getHiveTableNames(config.getHiveDatabaseForIntermediateTable());
+        Iterable<String> kylinIntermediates = Iterables.filter(hiveTableNames, new
Predicate<String>() {
+            @Override
+            public boolean apply(@Nullable String input) {
+                return input != null && input.startsWith("kylin_intermediate_");
+            }
+        });
 
-        String outputStr = result.getSecond();
-        BufferedReader reader = new BufferedReader(new StringReader(outputStr));
-        String line = null;
         List<String> allJobs = executableManager.getAllJobIds();
         List<String> allHiveTablesNeedToBeDeleted = new ArrayList<String>();
         List<String> workingJobList = new ArrayList<String>();
@@ -280,15 +281,14 @@ public class StorageCleanupJob extends AbstractApplication {
         }
         logger.info("Working jobIDs: " + workingJobList);
 
-        while ((line = reader.readLine()) != null) {
-
+        for (String line : kylinIntermediates) {
             logger.info("Checking table " + line);
 
             if (!line.startsWith(preFix))
                 continue;
 
             if (force == true) {
-                logger.warn("!!!!!!!!!!!!!!!Warning: will delete all intermediate hive tables!!!!!!!!!!!!!!!!!!!!!!");
+                logger.warn("Warning: will delete all intermediate hive tables!!!!!!!!!!!!!!!!!!!!!!");
                 allHiveTablesNeedToBeDeleted.add(line);
                 continue;
             }
@@ -320,7 +320,8 @@ public class StorageCleanupJob extends AbstractApplication {
         }
 
         if (delete == true) {
-            hiveCmdBuilder.reset();
+            final String useDatabaseHql = "USE " + config.getHiveDatabaseForIntermediateTable()
+ ";";
+            final HiveCmdBuilder hiveCmdBuilder = new HiveCmdBuilder();
             hiveCmdBuilder.addStatement(useDatabaseHql);
             for (String delHive : allHiveTablesNeedToBeDeleted) {
                 hiveCmdBuilder.addStatement("drop table if exists " + delHive + "; ");
@@ -339,9 +340,6 @@ public class StorageCleanupJob extends AbstractApplication {
             }
             System.out.println("----------------------------------------------------");
         }
-
-        if (reader != null)
-            reader.close();
     }
 
     private boolean isTableInUse(String segUuid, List<String> workingJobList) {


Mime
View raw message