Bladeren bron

HUE-564. Improve handling of beeswax active queries and expiration

Make RUNNING_QUERY_LIFETIME configurable.

Make the fields of driver null, only the ones not needed once execution has
finished. This will make the active queries light on memory while
they hang around before expiration.
Shrijeet Paliwal 14 jaren geleden
bovenliggende
commit
ca548cf292

+ 40 - 7
apps/beeswax/java/src/main/java/com/cloudera/beeswax/BeeswaxServiceImpl.java

@@ -90,10 +90,13 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
 
   private String notifyUrl;
 
+  // lifetime of a running query.
+  private long queryLifetime;
+
   /** Mapping between configuration variable names and descriptions. */
   private ConfigDescriptions configDescriptions = ConfigDescriptions.get();
 
-  private static final long RUNNING_QUERY_LIFETIME = 7*24*60*60*1000;  // 1 week
+  public static final long RUNNING_QUERY_LIFETIME = 7*24*60*60*1000;  // 1 week
   private static final long EVICTION_INTERVAL = 3*60*60*1000;  // 3 hours
   private static final String NOTIFY_URL_BASE = "/beeswax/query_cb/done/";
 
@@ -316,6 +319,12 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
           }
         }
       } finally {
+        // driver.plan.inputs and driver.plan.roottasks contains lots of
+        // pointers to memory; nullify them to allow for garbage collection.
+        synchronized (this) {
+          driver.getPlan().getInputs().clear();
+          driver.getPlan().getRootTasks().clear();
+        }
         notifyDone(this);
       }
     }
@@ -557,17 +566,31 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
     }
   }
 
+  /**
+   * Create a new BeeswaxServiceImpl with default query lifetime.
+   *
+   * @param dtHost The Hue host (ip or hostname).
+   * @param dtPort The port Desktop runs on.
+   * @param dtHttps Whether Desktop is running https.
+   */
+  public BeeswaxServiceImpl(String dtHost, int dtPort, boolean dtHttps) {
+    this(dtHost, dtPort, dtHttps, RUNNING_QUERY_LIFETIME);
+  }
 
   /**
    * Create a new BeeswaxServiceImpl.
-   * @param dtHost  The Hue host (ip or hostname).
-   * @param dtPort  The port Desktop runs on.
-   * @param dtHttps  Whether Desktop is running https.
+   *
+   * @param dtHost The Hue host (ip or hostname).
+   * @param dtPort The port Desktop runs on.
+   * @param dtHttps Whether Desktop is running https.
+   * @param queryLifetime The life time of a cached query.
    */
-  public BeeswaxServiceImpl(String dtHost, int dtPort, boolean dtHttps) {
+  public BeeswaxServiceImpl(String dtHost, int dtPort, boolean dtHttps,
+      long queryLifetime) {
     LogContext.initLogCapture();
     this.executor = Executors.newCachedThreadPool(new NamingThreadFactory("Beeswax-%d"));
     this.runningQueries = new ConcurrentHashMap<String, RunningQueryState>();
+    this.queryLifetime = queryLifetime;
 
     String protocol;
     if (dtHttps) {
@@ -599,7 +622,9 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
             long now = System.currentTimeMillis();
             for (Map.Entry<String, RunningQueryState> entry : runningQueries.entrySet()) {
               RunningQueryState rqState = entry.getValue();
-              if (rqState.getAtime() + RUNNING_QUERY_LIFETIME < now) {
+              //safe guard against small value of lifetime, only clean FINISHED or EXCEPTION state
+              if ((rqState.state == QueryState.FINISHED || rqState.state == QueryState.EXCEPTION )
+                    && rqState.getAtime() + getQueryLifetime() < now) {
                 String id = entry.getKey();
                 runningQueries.remove(id);
                 LOG.debug("Removed " + rqState.toString());
@@ -607,7 +632,7 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
               }
             }
 
-            LogContext.garbageCollect(RUNNING_QUERY_LIFETIME);
+            LogContext.garbageCollect(getQueryLifetime());
 
             long wakeup = now + EVICTION_INTERVAL;
             while (System.currentTimeMillis() < wakeup) {
@@ -876,4 +901,12 @@ public class BeeswaxServiceImpl implements BeeswaxService.Iface {
     }
     return ret;
   }
+
+  public long getQueryLifetime() {
+    return queryLifetime;
+  }
+
+  public void setQueryLifetime(long queryLifetime) {
+    this.queryLifetime = queryLifetime;
+  }
 }

+ 12 - 2
apps/beeswax/java/src/main/java/com/cloudera/beeswax/Server.java

@@ -58,6 +58,7 @@ public class Server {
   private static String dtHost = "";
   private static int dtPort = -1;
   private static boolean dtHttps = false;
+  private static long qlifetime = BeeswaxServiceImpl.RUNNING_QUERY_LIFETIME;
 
   /**
    * Parse command line options.
@@ -72,6 +73,11 @@ public class Server {
     metastoreOpt.setRequired(false);
     options.addOption(metastoreOpt);
 
+    Option queryLifetimeOpt = new Option("l", "query-lifetime", true,
+        "query lifetime");
+    queryLifetimeOpt.setRequired(false);
+    options.addOption(queryLifetimeOpt);
+
     Option beeswaxOpt = new Option("b", "beeswax", true, "port to use for beeswax");
     beeswaxOpt.setRequired(true);
     options.addOption(beeswaxOpt);
@@ -111,6 +117,8 @@ public class Server {
         dtPort = parsePort(opt);
       } else if (opt.getOpt() == "s") {
         dtHttps = true;
+      } else if (opt.getOpt() == "l") {
+        qlifetime = Long.valueOf(opt.getValue());
       }
     }
   }
@@ -185,14 +193,16 @@ public class Server {
    */
   private static void serveBeeswax(int port) throws TTransportException {
     TServerTransport serverTransport = new TServerSocket(port);
-    BeeswaxService.Iface impl = new BeeswaxServiceImpl(dtHost, dtPort, dtHttps);
+    BeeswaxService.Iface impl = new BeeswaxServiceImpl(dtHost, dtPort, dtHttps,
+        qlifetime);
     Processor processor = new BeeswaxService.Processor(impl);
     TThreadPoolServer.Options options = new TThreadPoolServer.Options();
     TServer server = new TThreadPoolServer(processor, serverTransport,
         new TTransportFactory(), new TTransportFactory(),
         new TBinaryProtocol.Factory(), new TBinaryProtocol.Factory(), options);
     LOG.info("Starting beeswax server on port " + port + ", talking back to Desktop at " +
-             dtHost + ":" + dtPort);
+             dtHost + ":" + dtPort + ", lifetime of queries set to " + qlifetime);
+
     server.serve();
   }
 

+ 6 - 0
apps/beeswax/src/beeswax/conf.py

@@ -84,3 +84,9 @@ METASTORE_CONN_TIMEOUT= Config(
   default=10,
   type=int,
   help='Timeouts in seconds for thrift calls to the hive metastore. This timeout should take into account that the metastore could talk to an external DB')
+
+BEESWAX_RUNNING_QUERY_LIFETIME = Config(
+  key='beeswax_running_query_lifetime',
+  default=604800000L, # 7*24*60*60*1000 (1 week)
+  type=long,
+  help='Time in seconds for beeswax to persist queries in its cache.')

+ 2 - 0
apps/beeswax/src/beeswax/management/commands/beeswax_server.py

@@ -57,6 +57,8 @@ class Command(NoArgsCommand):
       str(dt_host),
       '--desktop-port',
       str(desktop.conf.HTTP_PORT.get()),
+      '--query-lifetime',
+      str(beeswax.conf.BEESWAX_RUNNING_QUERY_LIFETIME.get()),
     ]
 
     # Running on HTTPS?