diff --git a/scalding/pom.xml b/scalding/pom.xml
index d77264c600..a5aae18f7e 100644
--- a/scalding/pom.xml
+++ b/scalding/pom.xml
@@ -33,6 +33,18 @@
   <name>Zeppelin: Scalding interpreter</name>
   <url>http://zeppelin.incubator.apache.org</url>
 
+  <properties>
+    <scala.version>2.10.5</scala.version>
+  </properties>
+
+  <repositories>
+    <repository>
+      <id>conjars</id>
+      <name>Concurrent Maven Repo</name>
+      <url>http://conjars.org/repo</url>
+    </repository>
+  </repositories>
+
   <dependencies>
     <dependency>
       <groupId>${project.groupId}</groupId>
@@ -62,6 +74,36 @@
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
+
+    <dependency>
+      <groupId>com.twitter</groupId>
+      <artifactId>scalding-core_2.10</artifactId>
+      <version>0.15.1-RC13</version>
+    </dependency>
+
+    <dependency>
+      <groupId>com.twitter</groupId>
+      <artifactId>scalding-repl_2.10</artifactId>
+      <version>0.15.1-RC13</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.scala-lang</groupId>
+      <artifactId>scala-library</artifactId>
+      <version>${scala.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.scala-lang</groupId>
+      <artifactId>scala-compiler</artifactId>
+      <version>${scala.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.scala-lang</groupId>
+      <artifactId>scala-reflect</artifactId>
+      <version>${scala.version}</version>
+    </dependency>
   </dependencies>
 
   <build>
@@ -128,6 +170,33 @@
           </execution>
         </executions>
       </plugin>
+      <!-- Plugin to compile Scala code -->
+      <plugin>
+        <groupId>org.scala-tools</groupId>
+        <artifactId>maven-scala-plugin</artifactId>
+        <executions>
+          <execution>
+            <id>compile</id>
+            <goals>
+              <goal>compile</goal>
+            </goals>
+            <phase>compile</phase>
+          </execution>
+          <execution>
+            <id>test-compile</id>
+            <goals>
+              <goal>testCompile</goal>
+            </goals>
+            <phase>test-compile</phase>
+          </execution>
+          <execution>
+            <phase>process-resources</phase>
+            <goals>
+              <goal>compile</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
     </plugins>
   </build>
 
diff --git a/scalding/src/main/java/org/apache/zeppelin/scalding/ScaldingInterpreter.java b/scalding/src/main/java/org/apache/zeppelin/scalding/ScaldingInterpreter.java
index bec52bc764..86bb801ef8 100644
--- a/scalding/src/main/java/org/apache/zeppelin/scalding/ScaldingInterpreter.java
+++ b/scalding/src/main/java/org/apache/zeppelin/scalding/ScaldingInterpreter.java
@@ -17,8 +17,14 @@
 
 package org.apache.zeppelin.scalding;
 
+import java.io.File;
 import java.util.Properties;
 import java.util.List;
+import java.util.LinkedList;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
 
 import org.apache.zeppelin.interpreter.Interpreter;
 import org.apache.zeppelin.interpreter.InterpreterContext;
@@ -29,8 +35,15 @@ import org.apache.zeppelin.scheduler.SchedulerFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.twitter.scalding.ScaldingILoop;
+
+import scala.Some;
+import scala.tools.nsc.Settings;
+import scala.tools.nsc.settings.MutableSettings.BooleanSetting;
+import scala.tools.nsc.settings.MutableSettings.PathSetting;
+
 /**
- * Scalding interpreter for Zeppelin.
+ * Scalding interpreter for Zeppelin. Based off the Spark interpreter code.
  *
  * @author sriramkrishnan
  *
@@ -42,12 +55,96 @@ public class ScaldingInterpreter extends Interpreter {
     Interpreter.register("scalding", ScaldingInterpreter.class.getName());
   }
 
+  private ScaldingILoop interpreter;
+
   public ScaldingInterpreter(Properties property) {
     super(property);
   }
 
   @Override
-  public void open() {}
+  public void open() {
+    URL[] urls = getClassloaderUrls();
+
+    // Very nice discussion about how scala compiler handle classpath
+    // https://groups.google.com/forum/#!topic/scala-user/MlVwo2xCCI0
+
+    /*
+     * > val env = new nsc.Settings(errLogger) > env.usejavacp.value = true > val p = new
+     * Interpreter(env) > p.setContextClassLoader > Alternatively you can set the class path through
+     * nsc.Settings.classpath.
+     *
+     * >> val settings = new Settings() >> settings.usejavacp.value = true >>
+     * settings.classpath.value += File.pathSeparator + >> System.getProperty("java.class.path") >>
+     * val in = new Interpreter(settings) { >> override protected def parentClassLoader =
+     * getClass.getClassLoader >> } >> in.setContextClassLoader()
+     */
+    Settings settings = new Settings();
+
+    // set classpath for scala compiler
+    PathSetting pathSettings = settings.classpath();
+    String classpath = "";
+    List<File> paths = currentClassPath();
+    for (File f : paths) {
+      if (classpath.length() > 0) {
+        classpath += File.pathSeparator;
+      }
+      classpath += f.getAbsolutePath();
+    }
+
+    if (urls != null) {
+      for (URL u : urls) {
+        if (classpath.length() > 0) {
+          classpath += File.pathSeparator;
+        }
+        classpath += u.getFile();
+      }
+    }
+
+    pathSettings.v_$eq(classpath);
+    settings.scala$tools$nsc$settings$ScalaSettings$_setter_$classpath_$eq(pathSettings);
+
+
+    // set classloader for scala compiler
+    settings.explicitParentLoader_$eq(new Some<ClassLoader>(Thread.currentThread()
+        .getContextClassLoader()));
+    BooleanSetting b = (BooleanSetting) settings.usejavacp();
+    b.v_$eq(true);
+    settings.scala$tools$nsc$settings$StandardScalaSettings$_setter_$usejavacp_$eq(b);
+
+    /* Scalding interpreter */
+    interpreter = new ScaldingILoop();
+    interpreter.settings_$eq(settings);
+    interpreter.createInterpreter();
+  }
+
+  private List<File> currentClassPath() {
+    List<File> paths = classPath(Thread.currentThread().getContextClassLoader());
+    String[] cps = System.getProperty("java.class.path").split(File.pathSeparator);
+    if (cps != null) {
+      for (String cp : cps) {
+        paths.add(new File(cp));
+      }
+    }
+    return paths;
+  }
+
+  private List<File> classPath(ClassLoader cl) {
+    List<File> paths = new LinkedList<File>();
+    if (cl == null) {
+      return paths;
+    }
+
+    if (cl instanceof URLClassLoader) {
+      URLClassLoader ucl = (URLClassLoader) cl;
+      URL[] urls = ucl.getURLs();
+      if (urls != null) {
+        for (URL url : urls) {
+          paths.add(new File(url.getFile()));
+        }
+      }
+    }
+    return paths;
+  }
 
   @Override
   public void close() {}
@@ -55,9 +152,80 @@ public class ScaldingInterpreter extends Interpreter {
 
   @Override
   public InterpreterResult interpret(String cmd, InterpreterContext contextInterpreter) {
-    logger.debug("Run Scalding command '" + cmd + "'");
-    System.out.println("Running Scalding Interpreter!");
-    return new InterpreterResult(InterpreterResult.Code.SUCCESS, "Success!!");
+    logger.info("Run Scalding command '" + cmd + "'");
+
+    if (cmd == null || cmd.trim().length() == 0) {
+      return new InterpreterResult(Code.SUCCESS);
+    }
+    return interpret(cmd.split("\n"), contextInterpreter);
+  }
+
+  public InterpreterResult interpret(String[] lines, InterpreterContext context) {
+    synchronized (this) {
+      InterpreterResult r = interpretInput(lines);
+      return r;
+    }
+  }
+
+  public InterpreterResult interpretInput(String[] lines) {
+
+    // add print("") to make sure not finishing with comment
+    // see https://github.com/NFLabs/zeppelin/issues/151
+    String[] linesToRun = new String[lines.length + 1];
+    for (int i = 0; i < lines.length; i++) {
+      linesToRun[i] = lines[i];
+    }
+    linesToRun[lines.length] = "print(\"\")";
+
+    Code r = null;
+    String incomplete = "";
+
+    for (int l = 0; l < linesToRun.length; l++) {
+      String s = linesToRun[l];
+      // check if next line starts with "." (but not ".." or "./") it is treated as an invocation
+      if (l + 1 < linesToRun.length) {
+        String nextLine = linesToRun[l + 1].trim();
+        if (nextLine.startsWith(".") && !nextLine.startsWith("..") && !nextLine.startsWith("./")) {
+          incomplete += s + "\n";
+          continue;
+        }
+      }
+
+      scala.tools.nsc.interpreter.Results.Result res = null;
+      try {
+        res = interpreter.intp().interpret(incomplete + s);
+      } catch (Exception e) {
+        e.printStackTrace();
+        logger.error("Interpreter exception: ", e);
+        return new InterpreterResult(Code.ERROR, e.getMessage());
+      }
+
+      r = getResultCode(res);
+
+      if (r == Code.ERROR) {
+        return new InterpreterResult(r);
+      } else if (r == Code.INCOMPLETE) {
+        incomplete += s + "\n";
+      } else {
+        incomplete = "";
+      }
+    }
+
+    if (r == Code.INCOMPLETE) {
+      return new InterpreterResult(r, "Incomplete expression");
+    } else {
+      return new InterpreterResult(r);
+    }
+  }
+
+  private Code getResultCode(scala.tools.nsc.interpreter.Results.Result r) {
+    if (r instanceof scala.tools.nsc.interpreter.Results.Success$) {
+      return Code.SUCCESS;
+    } else if (r instanceof scala.tools.nsc.interpreter.Results.Incomplete$) {
+      return Code.INCOMPLETE;
+    } else {
+      return Code.ERROR;
+    }
   }
 
   @Override