From 35fc032d7e6cc679cf3604f44cf6ba4a3545a53b Mon Sep 17 00:00:00 2001 From: Sriram Krishnan Date: Thu, 17 Dec 2015 13:22:52 -0800 Subject: [PATCH] Initial version of a ScaldingInterpreter running in local mode. Need to get console output next. And add tests, and make it work for HDFS. --- scalding/pom.xml | 69 +++++++ .../scalding/ScaldingInterpreter.java | 178 +++++++++++++++++- 2 files changed, 242 insertions(+), 5 deletions(-) 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 @@ Zeppelin: Scalding interpreter http://zeppelin.incubator.apache.org + + 2.10.5 + + + + + conjars + Concurrent Maven Repo + http://conjars.org/repo + + + ${project.groupId} @@ -62,6 +74,36 @@ junit test + + + com.twitter + scalding-core_2.10 + 0.15.1-RC13 + + + + com.twitter + scalding-repl_2.10 + 0.15.1-RC13 + + + + org.scala-lang + scala-library + ${scala.version} + + + + org.scala-lang + scala-compiler + ${scala.version} + + + + org.scala-lang + scala-reflect + ${scala.version} + @@ -128,6 +170,33 @@ + + + org.scala-tools + maven-scala-plugin + + + compile + + compile + + compile + + + test-compile + + testCompile + + test-compile + + + process-resources + + compile + + + + 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 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(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 currentClassPath() { + List 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 classPath(ClassLoader cl) { + List paths = new LinkedList(); + 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