diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
index edcf5133667..27ccfab66f9 100755
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
@@ -428,6 +428,7 @@ public static enum ConfVars {
     // Decide when new note is created, interpreter settings will be binded automatically or not.
     ZEPPELIN_NOTEBOOK_AUTO_INTERPRETER_BINDING("zeppelin.notebook.autoInterpreterBinding", true),
     ZEPPELIN_CONF_DIR("zeppelin.conf.dir", "conf"),
+    ZEPPELIN_INTERPRETER_REPO_DIR("zeppelin.user.interpreter.dir", "interpreter"),
     ZEPPELIN_DEP_LOCALREPO("zeppelin.dep.localrepo", "local-repo"),
     // Allows a way to specify a ',' separated list of allowed origins for rest and websockets
     // i.e. http://localhost:8080
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java
index 4ff0cc3ad59..f76de3d202d 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java
@@ -53,7 +53,7 @@ public class InterpreterFactory {
       .synchronizedMap(new HashMap<String, URLClassLoader>());
 
   private ZeppelinConfiguration conf;
-  String[] interpreterClassList;
+  List<String> interpreterClassList;
 
   private Map<String, InterpreterSetting> interpreterSettings =
       new HashMap<String, InterpreterSetting>();
@@ -75,7 +75,6 @@ public InterpreterFactory(ZeppelinConfiguration conf,
     this(conf, new InterpreterOption(true), angularObjectRegistryListener, depResolver);
   }
 
-
   public InterpreterFactory(ZeppelinConfiguration conf, InterpreterOption defaultOption,
       AngularObjectRegistryListener angularObjectRegistryListener,
       DependencyResolver depResolver)
@@ -85,8 +84,10 @@ public InterpreterFactory(ZeppelinConfiguration conf, InterpreterOption defaultO
     this.angularObjectRegistryListener = angularObjectRegistryListener;
     this.depResolver = depResolver;
     String replsConf = conf.getString(ConfVars.ZEPPELIN_INTERPRETERS);
-    interpreterClassList = replsConf.split(",");
-
+    interpreterClassList = new ArrayList<String>();
+    for (String className : replsConf.split(",")) {
+      interpreterClassList.add(className);
+    }
     GsonBuilder builder = new GsonBuilder();
     builder.setPrettyPrinting();
     builder.registerTypeAdapter(Interpreter.class, new InterpreterSerializer());
@@ -95,6 +96,96 @@ public InterpreterFactory(ZeppelinConfiguration conf, InterpreterOption defaultO
     init();
   }
 
+  public boolean loadDynamicInterpreter(String intpGroupName, String intpName, String artifact,
+                                        String intpClassName) {
+    return loadDynamicInterpreter(intpGroupName, intpName, artifact, intpClassName, null, false);
+  }
+
+  public boolean loadDynamicInterpreter(String intpGroupName, String intpName, String artifact,
+                                        String intpClassName, String repositoryUrl,
+                                        boolean isSnapShotRepo) {
+    String[] artifactItem = artifact.split(":");
+    String interpreterDesPath = conf.getString(ConfVars.ZEPPELIN_INTERPRETER_REPO_DIR);
+    interpreterDesPath += "/" + intpGroupName + "/" + intpName + "/";
+
+    String interpreterLoadPath = conf.getRelativeDir(ConfVars.ZEPPELIN_INTERPRETER_REPO_DIR);
+    interpreterLoadPath += "/" + intpGroupName + "/" + intpName;
+
+    if (artifactItem.length <= 0) {
+      logger.error("invalid artifact [" + artifact + "]");
+      return false;
+    }
+
+    try {
+      if (repositoryUrl != null) {
+        depResolver.addRepo("dyInterpreterRepo", repositoryUrl, isSnapShotRepo);
+      }
+      logger.info("interpreter- path {}", interpreterLoadPath);
+      depResolver.load(artifact, interpreterDesPath);
+      setDynamicInterpreter(intpClassName, interpreterLoadPath);
+    } catch (Exception e) {
+      logger.error(e.getMessage());
+      return false;
+    }
+    return true;
+  }
+
+  public boolean unloadDynamicInterpreter(String intpGorupName, String intpName) {
+    try {
+      remove(intpName);
+    } catch (Exception e) {
+      logger.error(e.getMessage());
+      return false;
+    }
+    return true;
+  }
+
+  protected void setDynamicInterpreter(String interpreterClassName, String fileDirPath)
+          throws InterpreterException, IOException {
+    logger.info("loadDynamicInterpreter !!");
+
+    ClassLoader oldcl = Thread.currentThread().getContextClassLoader();
+    interpreterClassList.add(interpreterClassName);
+    // Load classes
+    File interpreterDir = new File(fileDirPath);
+
+    if (interpreterDir != null) {
+      URL[] urls = null;
+      try {
+        urls = recursiveBuildLibList(interpreterDir);
+      } catch (MalformedURLException e1) {
+        logger.error("Can't load jars ", e1);
+      }
+      URLClassLoader ccl = new URLClassLoader(urls, oldcl);
+
+      logger.info("load dynamic interpreter [" + interpreterClassName + "]");
+
+      try {
+        Class.forName(interpreterClassName, true, ccl);
+        Set<String> keys = Interpreter.registeredInterpreters.keySet();
+        for (String intName : keys) {
+          if (interpreterClassName.equals(
+                  Interpreter.registeredInterpreters.get(intName).getClassName())) {
+            Interpreter.registeredInterpreters.get(intName).setPath(fileDirPath);
+            logger.info("Interpreter " + intName + " found. class=" + fileDirPath);
+            cleanCl.put(fileDirPath, ccl);
+          }
+        }
+      } catch (ClassNotFoundException e) {
+        logger.error("Load error {}", e);
+      }
+    }
+
+    for (String settingId : interpreterSettings.keySet()) {
+      InterpreterSetting setting = interpreterSettings.get(settingId);
+      logger.info("Interpreter setting group {} : id={}, name={}",
+              setting.getGroup(), settingId, setting.getName());
+      for (Interpreter interpreter : setting.getInterpreterGroup()) {
+        logger.info("  className = {}", interpreter.getClassName());
+      }
+    }
+  }
+
   private void init() throws InterpreterException, IOException {
     ClassLoader oldcl = Thread.currentThread().getContextClassLoader();