Adding more configurations to livy interpreter

mfelgamal · mfelgamal · commit 4c0d5f06e9b6 · 2016-06-01T10:29:27.000+02:00
diff --git a/livy/src/main/java/org/apache/zeppelin/livy/LivyHelper.java b/livy/src/main/java/org/apache/zeppelin/livy/LivyHelper.java
@@ -60,27 +60,45 @@ public class LivyHelper {
 
   public Integer createSession(InterpreterContext context, String kind) throws Exception {
     try {
+      Map<String, String> conf = new HashMap<String, String>();
+      
+      conf.put("spark.master", property.getProperty("zeppelin.livy.master"));
+      
+      conf.put("spark.driver.cores", property.getProperty("spark.driver.cores"));
+      conf.put("spark.executor.cores", property.getProperty("spark.executor.cores"));
+      conf.put("spark.driver.memory", property.getProperty("spark.driver.memory"));
+      conf.put("spark.executor.memory", property.getProperty("spark.executor.memory"));
+
+      if (!property.getProperty("spark.dynamicAllocation.enabled").equals("true")) {
+        conf.put("spark.executor.instances", property.getProperty("spark.executor.instances"));
+      }
+
+      if (property.getProperty("spark.dynamicAllocation.enabled").equals("true")) {
+        conf.put("spark.dynamicAllocation.enabled",
+            property.getProperty("spark.dynamicAllocation.enabled"));
+        conf.put("spark.shuffle.service.enabled", "true");
+        conf.put("spark.dynamicAllocation.cachedExecutorIdleTimeout",
+            property.getProperty("spark.dynamicAllocation.cachedExecutorIdleTimeout"));
+        conf.put("spark.dynamicAllocation.minExecutors",
+            property.getProperty("spark.dynamicAllocation.minExecutors"));
+        conf.put("spark.dynamicAllocation.initialExecutors",
+            property.getProperty("spark.dynamicAllocation.initialExecutors"));
+        conf.put("spark.dynamicAllocation.maxExecutors",
+            property.getProperty("spark.dynamicAllocation.maxExecutors"));
+      }
+
+      String confData = gson.toJson(conf);
+
       String json = executeHTTP(property.getProperty("zeppelin.livy.url") + "/sessions",
-          "POST",
+          "POST", 
           "{" +
               "\"kind\": \"" + kind + "\", " +
-              "\"master\": \"" + property.getProperty("zeppelin.livy.master") + "\", " +
-              "\"proxyUser\": \"" + context.getAuthenticationInfo().getUser() + "\"" +
+              "\"conf\": " + confData + ", " + 
+              "\"proxyUser\": " + context.getAuthenticationInfo().getUser() + 
               "}",
           context.getParagraphId()
       );
-      if (json.contains("CreateInteractiveRequest[\\\"master\\\"]")) {
-        json = executeHTTP(property.getProperty("zeppelin.livy.url") + "/sessions",
-            "POST",
-            "{" +
-                "\"kind\": \"" + kind + "\", " +
-                "\"conf\":{\"spark.master\": \""
-                + property.getProperty("zeppelin.livy.master") + "\"}," +
-                "\"proxyUser\": \"" + context.getAuthenticationInfo().getUser() + "\"" +
-                "}",
-            context.getParagraphId()
-        );
-      }
+
       Map jsonMap = (Map<Object, Object>) gson.fromJson(json,
           new TypeToken<Map<Object, Object>>() {
           }.getType());
diff --git a/livy/src/main/java/org/apache/zeppelin/livy/LivySparkInterpreter.java b/livy/src/main/java/org/apache/zeppelin/livy/LivySparkInterpreter.java
@@ -46,6 +46,21 @@ public class LivySparkInterpreter extends Interpreter {
         new InterpreterPropertyBuilder()
             .add("zeppelin.livy.url", DEFAULT_URL, "The URL for Livy Server.")
             .add("zeppelin.livy.master", LOCAL, "Spark master uri. ex) spark://masterhost:7077")
+            .add("spark.driver.cores", "1", "Driver cores. ex) 1, 2")
+            .add("spark.driver.memory", "512m", "Driver memory. ex) 512m, 32g")
+            .add("spark.executor.instances", "3", "Executor instances. ex) 1, 4")
+            .add("spark.executor.cores", "1", "Num cores per executor. ex) 1, 4")
+            .add("spark.executor.memory", "512m", 
+                 "Executor memory per worker instance. ex) 512m, 32g")
+            .add("spark.dynamicAllocation.enabled", "false", "Use dynamic resource allocation")
+            .add("spark.dynamicAllocation.cachedExecutorIdleTimeout", "120s", 
+                 "Remove an executor which has cached data blocks")
+            .add("spark.dynamicAllocation.minExecutors", "0", 
+                 "Lower bound for the number of executors if dynamic allocation is enabled. ")
+            .add("spark.dynamicAllocation.initialExecutors", "1", 
+                 "Initial number of executors to run if dynamic allocation is enabled. ")
+            .add("spark.dynamicAllocation.maxExecutors", "10", 
+                 "Upper bound for the number of executors if dynamic allocation is enabled. ")
             .build()
     );
   }
