Browse Source

[spark] Refactor and add initial repl that captures the input/output of spark

Erick Tryzelaar 11 năm trước cách đây
mục cha
commit
a20a7ba1a5

+ 0 - 39
apps/spark/java/pom.xml

@@ -49,12 +49,6 @@
 
     <properties>
         <javaVersion>1.7</javaVersion>
-        <!--
-        <hadoop.version>${cdh.hadoop.version}</hadoop.version>
-        <spark.version>${cdh.spark.version}</spark.version>
-        <slf4j.version>${cdh.slf4j.version}</slf4j.version>
-        <jetty.version>8.1.14.v20131031</jetty.version>
-        -->
     </properties>
 
     <modules>
@@ -240,39 +234,6 @@
             </plugin>
             -->
 
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-dependency-plugin</artifactId>
-                <executions>
-                    <execution>
-                        <id>copy-dependencies</id>
-                        <phase>prepare-package</phase>
-                        <goals>
-                            <goal>copy-dependencies</goal>
-                        </goals>
-                        <configuration>
-                            <outputDirectory>${project.build.directory}/lib</outputDirectory>
-                            <overWriteReleases>false</overWriteReleases>
-                            <overWriteSnapshots>false</overWriteSnapshots>
-                            <overWriteIfNewer>true</overWriteIfNewer>
-                        </configuration>
-                    </execution>
-                </executions>
-            </plugin>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-jar-plugin</artifactId>
-                <configuration>
-                    <archive>
-                        <manifest>
-                            <addClasspath>true</addClasspath>
-                            <classpathPrefix>lib/</classpathPrefix>
-                            <mainClass>com.cloudera.sparker.SparkerMain</mainClass>
-                        </manifest>
-                    </archive>
-                </configuration>
-            </plugin>
-
         </plugins>
 
     </build>

+ 125 - 0
apps/spark/java/sparker-repl/pom.xml

@@ -14,7 +14,12 @@
     <packaging>jar</packaging>
 
     <properties>
+        <scala.version>2.10.4</scala.version>
+        <scala.binary.version>2.10</scala.binary.version>
+        <scala.macros.version>2.0.1</scala.macros.version>
         <spark.version>1.1.0-cdh5.2.0-SNAPSHOT</spark.version>
+        <PermGen>64m</PermGen>
+        <MaxPermGen>512m</MaxPermGen>
     </properties>
 
     <dependencies>
@@ -27,4 +32,124 @@
 
     </dependencies>
 
+    <build>
+        <plugins>
+
+            <plugin>
+                <groupId>org.scala-tools</groupId>
+                <artifactId>maven-scala-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>compile</goal>
+                            <goal>testCompile</goal>
+                        </goals>
+                    </execution>
+                    <!--
+                    <execution>
+                        <id>scala-compile-first</id>
+                        <phase>process-resources</phase>
+                        <goals>
+                            <goal>compile</goal>
+                        </goals>
+                    </execution>
+                    <execution>
+                        <id>scala-test-compile-first</id>
+                        <phase>process-test-resources</phase>
+                        <goals>
+                            <goal>testCompile</goal>
+                        </goals>
+                    </execution>
+                    <execution>
+                        <id>attach-scaladocs</id>
+                        <phase>verify</phase>
+                        <goals>
+                            <goal>doc-jar</goal>
+                        </goals>
+                    </execution>
+                    -->
+                </executions>
+                <configuration>
+                    <!--
+                    <scalaVersion>${scala.version}</scalaVersion>
+                    <recompileMode>incremental</recompileMode>
+                    <useZincServer>true</useZincServer>
+                    <args>
+                        <arg>-unchecked</arg>
+                        <arg>-deprecation</arg>
+                        <arg>-feature</arg>
+                        <arg>-language:postfixOps</arg>
+                    </args>
+                    <jvmArgs>
+                        <jvmArg>-Xms1024m</jvmArg>
+                        <jvmArg>-Xmx1024m</jvmArg>
+                        <jvmArg>-XX:PermSize=${PermGen}</jvmArg>
+                        <jvmArg>-XX:MaxPermSize=${MaxPermGen}</jvmArg>
+                    </jvmArgs>
+                    <javacArgs>
+                        <javacArg>-source</javacArg>
+                        <javacArg>${java.version}</javacArg>
+                        <javacArg>-target</javacArg>
+                        <javacArg>${java.version}</javacArg>
+                    </javacArgs>
+                    -->
+                    <!-- The following plugin is required to use quasiquotes in Scala 2.10 and is used
+                         by Spark SQL for code generation. -->
+                    <compilerPlugins>
+                        <compilerPlugin>
+                            <groupId>org.scalamacros</groupId>
+                            <artifactId>paradise_${scala.version}</artifactId>
+                            <version>${scala.macros.version}</version>
+                        </compilerPlugin>
+                    </compilerPlugins>
+                </configuration>
+            </plugin>
+
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-dependency-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>copy-dependencies</id>
+                        <phase>prepare-package</phase>
+                        <goals>
+                            <goal>copy-dependencies</goal>
+                        </goals>
+                        <configuration>
+                            <outputDirectory>${project.build.directory}/lib</outputDirectory>
+                            <overWriteReleases>false</overWriteReleases>
+                            <overWriteSnapshots>false</overWriteSnapshots>
+                            <overWriteIfNewer>true</overWriteIfNewer>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <configuration>
+                    <archive>
+                        <manifest>
+                            <addClasspath>true</addClasspath>
+                            <classpathPrefix>lib/</classpathPrefix>
+                            <mainClass>com.cloudera.hue.sparker.repl.Main</mainClass>
+                        </manifest>
+                    </archive>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+    <reporting>
+        <plugins>
+            <plugin>
+                <groupId>org.scala-tools</groupId>
+                <artifactId>maven-scala-plugin</artifactId>
+                <configuration>
+                    <scalaVersion>${scala.version}</scalaVersion>
+                </configuration>
+            </plugin>
+        </plugins>
+    </reporting>
+
 </project>

+ 1 - 1
apps/spark/java/sparker-repl/src/main/java/com/cloudera/hue/sparker/SparkerInterpreter.java → apps/spark/java/sparker-repl/src/main/java/com/cloudera/hue/sparker/repl/SparkerInterpreter.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package com.cloudera.hue.sparker;
+package com.cloudera.hue.sparker.repl;
 
 import org.apache.spark.repl.Main;
 import org.apache.spark.repl.SparkILoop;

+ 10 - 0
apps/spark/java/sparker-repl/src/main/scala/com/cloudera/hue/sparker/repl/Main.scala

@@ -0,0 +1,10 @@
+package com.cloudera.hue.sparker.repl
+
+import java.io.StringWriter
+
+object Main {
+  def main(args: Array[String]): Unit = {
+    org.apache.spark.repl.Main.interp = new SparkerILoop(Console.in, new StringWriter)
+    org.apache.spark.repl.Main.interp.process(args)
+  }
+}

+ 107 - 0
apps/spark/java/sparker-repl/src/main/scala/com/cloudera/hue/sparker/repl/SparkerILoop.scala

@@ -0,0 +1,107 @@
+package com.cloudera.hue.sparker.repl
+
+import java.io.{BufferedReader, StringWriter}
+
+import scala.tools.nsc.interpreter._
+
+import org.apache.spark.repl.{SparkIMain, SparkILoop}
+
+import scala.tools.nsc.SparkHelper
+import scala.tools.nsc.interpreter.Formatting
+import scala.tools.nsc.util.ClassPath
+
+class SparkerILoop(in0: BufferedReader, outString: StringWriter) extends SparkILoop(in0, new JPrintWriter(outString)) {
+
+  class SparkerILoopInterpreter extends SparkILoopInterpreter {
+    outer =>
+
+    override lazy val formatting = new Formatting {
+      def prompt = SparkerILoop.this.prompt
+    }
+    override protected def parentClassLoader = SparkHelper.explicitParentLoader(settings).getOrElse(classOf[SparkILoop].getClassLoader)
+
+    override def interpret(line: String, synthetic: Boolean): IR.Result = {
+      val result = super.interpret(line, synthetic)
+        /*
+      match result {
+        case IR.Success(foo) =>
+      }
+      */
+      print("interpret: " + result + "\n")
+      result
+    }
+  }
+
+  /** Create a new interpreter. */
+  override def createInterpreter() {
+    require(settings != null)
+
+    if (addedClasspath != "") settings.classpath.append(addedClasspath)
+    // work around for Scala bug
+    val totalClassPath = SparkILoop.getAddedJars.foldLeft(
+      settings.classpath.value)((l, r) => ClassPath.join(l, r))
+    this.settings.classpath.value = totalClassPath
+
+    intp = new SparkerILoopInterpreter
+  }
+
+  private val replayQuestionMessage =
+    """|That entry seems to have slain the compiler.  Shall I replay
+      |your session? I can re-run each line except the last one.
+      |[y/n]
+    """.trim.stripMargin
+
+  private def crashRecovery(ex: Throwable): Boolean = {
+    echo(ex.toString)
+    ex match {
+      case _: NoSuchMethodError | _: NoClassDefFoundError =>
+        echo("\nUnrecoverable error.")
+        throw ex
+      case _  =>
+        def fn(): Boolean =
+          try in.readYesOrNo(replayQuestionMessage, { echo("\nYou must enter y or n.") ; fn() })
+          catch { case _: RuntimeException => false }
+
+        if (fn()) replay()
+        else echo("\nAbandoning crashed session.")
+    }
+    true
+  }
+
+  override def loop(): Unit = {
+    def readOneLine() = {
+      out.flush()
+      val line = in readLine prompt
+      print("readOneLine: " + line + "\n")
+      line
+    }
+    // return false if repl should exit
+    def processLine(line: String): Boolean = {
+      print("processLine: " + line + "\n")
+      if (isAsync) {
+        if (!awaitInitialized()) return false
+        runThunks()
+      }
+      if (line eq null) false               // assume null means EOF
+      else command(line) match {
+        case Result(false, _)           => false
+        case Result(_, Some(finalLine)) => {
+          print("out: " + finalLine)
+          print("out2: (" + outString.getBuffer.toString + ")")
+          addReplay(finalLine)
+        } ; true
+        case _                          => true
+      }
+    }
+    def innerLoop() {
+      outString.getBuffer.setLength(0)
+      val shouldContinue = try {
+        processLine(readOneLine())
+      } catch {case t: Throwable => crashRecovery(t)}
+      if (shouldContinue)
+        innerLoop()
+    }
+    innerLoop()
+  }
+
+}

+ 6 - 4
apps/spark/java/sparker-server/pom.xml

@@ -30,9 +30,6 @@
     <artifactId>sparker-server</artifactId>
     <packaging>jar</packaging>
 
-    <name>Spark Server</name>
-    <description>Spark Server</description>
-
     <dependencies>
         <dependency>
             <groupId>org.eclipse.jetty</groupId>
@@ -49,6 +46,11 @@
             <artifactId>jackson-mapper-asl</artifactId>
             <version>1.9.3</version>
         </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>14.0.1</version>
+        </dependency>
     </dependencies>
 
     <build>
@@ -80,7 +82,7 @@
                         <manifest>
                             <addClasspath>true</addClasspath>
                             <classpathPrefix>lib/</classpathPrefix>
-                            <mainClass>com.cloudera.hue.sparker.SparkerMain</mainClass>
+                            <mainClass>com.cloudera.hue.sparker.server.SparkerMain</mainClass>
                         </manifest>
                     </archive>
                 </configuration>

+ 1 - 1
apps/spark/java/sparker-server/src/main/java/com/cloudera/hue/sparker/Session.java → apps/spark/java/sparker-server/src/main/java/com/cloudera/hue/sparker/server/Session.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package com.cloudera.hue.sparker;
+package com.cloudera.hue.sparker.server;
 
 import java.io.IOException;
 import java.util.List;

+ 1 - 1
apps/spark/java/sparker-server/src/main/java/com/cloudera/hue/sparker/SessionManager.java → apps/spark/java/sparker-server/src/main/java/com/cloudera/hue/sparker/server/SessionManager.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package com.cloudera.hue.sparker;
+package com.cloudera.hue.sparker.server;
 
 import java.io.IOException;
 import java.util.Enumeration;

+ 1 - 1
apps/spark/java/sparker-server/src/main/java/com/cloudera/hue/sparker/SparkerMain.java → apps/spark/java/sparker-server/src/main/java/com/cloudera/hue/sparker/server/SparkerMain.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package com.cloudera.hue.sparker;
+package com.cloudera.hue.sparker.server;
 
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.servlet.ServletContextHandler;

+ 1 - 1
apps/spark/java/sparker-server/src/main/java/com/cloudera/hue/sparker/SparkerServlet.java → apps/spark/java/sparker-server/src/main/java/com/cloudera/hue/sparker/server/SparkerServlet.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package com.cloudera.hue.sparker;
+package com.cloudera.hue.sparker.server;
 
 import org.codehaus.jackson.map.ObjectMapper;
 import org.codehaus.jackson.map.ObjectWriter;

+ 1 - 1
apps/spark/java/sparker-server/src/main/java/com/cloudera/hue/sparker/SparkerSession.java → apps/spark/java/sparker-server/src/main/java/com/cloudera/hue/sparker/server/SparkerSession.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package com.cloudera.hue.sparker;
+package com.cloudera.hue.sparker.server;
 
 import com.google.common.collect.Lists;
 

+ 1 - 1
apps/spark/spark_server.sh

@@ -47,7 +47,7 @@ set -o errexit
 #echo \$HIVE_HOME=$HIVE_HOME
 #
 SPARK_ROOT=$(dirname $0)
-SPARK_JAR=$SPARK_ROOT/java/target/spark-server-3.7.0-SNAPSHOT.jar
+SPARK_JAR=$SPARK_ROOT/java/sparker-server/target/sparker-server-3.7.0-SNAPSHOT.jar
 #HIVE_LIB=$HIVE_HOME/lib
 #
 #export HADOOP_CLASSPATH=$(find $HADOOP_HOME -name hue-plugins*.jar | tr "\n" :):$(find $HIVE_LIB -name "*.jar" | tr "\n" :)