1

I am trying to dinamically interpret code given as a String in Spark, just like calling the eval in Perl language. However, I got problem when running the program. Really appreciate for your help.

Requirement:

The requirement is to make the spark processing chain configurable. For example, customer could set the processing steps in configuration file as below. Steps:

1) textFile("files///<file_full_path>") 
2) flatMap(line => line.split(" ")) 
3) map(word => (word, 1)) 
4) reduceByKey(_ + _) 
5) foreach(println)

All above steps are defined in a configuration file. Then, the spark driver will load the configuration file and make the processing steps as a string, such as:

val processFlow = 
"""
sc.textFile("file:///input.txt").flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey(_ + _).foreach(println)
"""

Then, Spark will execute the piece of code defined in above variable processFlow.

Here is my Spark source code:

It is from word count sample, I just make the RDD methods invoked by interpreter as a string.

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import scala.collection.mutable.{Map, ArraySeq}
import scala.tools.nsc.GenericRunnerSettings
import scala.tools.nsc.interpreter.IMain
class TestMain {
  def exec(): Unit = {
    val out = System.out
    val flusher = new java.io.PrintWriter(out)
    val interpreter = {
      val settings = new GenericRunnerSettings( println _ )
      settings.usejavacp.value = true
      new IMain(settings, flusher)
    }
    val conf = new SparkConf().setAppName("TestMain")
    val sc = new SparkContext(conf)
    val methodChain =
      """
      val textFile = sc.textFile("file:///input.txt")
      textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey(_ + _).foreach(println)
      """
    interpreter.bind("sc", sc);
    val resultFlag = interpreter.interpret(methodChain)
  }
}
object TestMain {
  def main(args: Array[String]) {
    val testMain = new TestMain()
    testMain.exec()
    System.exit(0)
  }
}

Problem:

However, I got an error when running above Spark code (master=local), logs as below.

sc: org.apache.spark.SparkContext = org.apache.spark.SparkContext@7d87addd
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 1 times, most recent failure: Lost task 0.0 in stage 0.0 (TID 0, localhost): java.lang.ClassNotFoundException: $anonfun$1
        at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
        at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
        at java.security.AccessController.doPrivileged(Native Method)
        at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
        at java.lang.Class.forName0(Native Method)
        at java.lang.Class.forName(Class.java:270)
        at org.apache.spark.serializer.JavaDeserializationStream$$anon$1.resolveClass(JavaSerializer.scala:68)
        at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1612)
        at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1517)
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1771)
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
        at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1990)
        at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1915)
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
        at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1990)
        at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1915)
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
        at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1990)
        at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1915)
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
        at java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
        at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:76)
        at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:115)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)
        at org.apache.spark.scheduler.Task.run(Task.scala:89)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
        at java.lang.Thread.run(Thread.java:745)

Driver stacktrace:
        at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1431)
        at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1419)
        at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1418)
        at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
        at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
        at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1418)
        at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:799)
        at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:799)
        at scala.Option.foreach(Option.scala:236)
        at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:799)
        at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1640)
        at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1599)
        at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1588)
        at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
        at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:620)
        at org.apache.spark.SparkContext.runJob(SparkContext.scala:1843)
        at org.apache.spark.SparkContext.runJob(SparkContext.scala:1856)
        at org.apache.spark.SparkContext.runJob(SparkContext.scala:1869)
        at org.apache.spark.SparkContext.runJob(SparkContext.scala:1940)
        at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:927)
        at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
        at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:111)
        at org.apache.spark.rdd.RDD.withScope(RDD.scala:316)
        at org.apache.spark.rdd.RDD.collect(RDD.scala:926)
        at .<init>(<console>:12)
        at .<clinit>(<console>)
        at .<init>(<console>:7)
        at .<clinit>(<console>)
        at $print(<console>)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:606)
        at scala.tools.nsc.interpreter.IMain$ReadEvalPrint.call(IMain.scala:734)
        at scala.tools.nsc.interpreter.IMain$Request.loadAndRun(IMain.scala:983)
        at scala.tools.nsc.interpreter.IMain.loadAndRunReq$1(IMain.scala:573)
        at scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:604)
        at scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:568)
        at com.tr.ecp.test.TestMain.exec(TestMain.scala:44)
        at com.tr.ecp.test.TestMain$.main(TestMain.scala:57)
        at com.tr.ecp.test.TestMain.main(TestMain.scala)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:606)
        at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:731)
        at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:181)
        at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:206)
        at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:121)
        at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.lang.ClassNotFoundException: $anonfun$1
        at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
        at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
        at java.security.AccessController.doPrivileged(Native Method)
        at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
        at java.lang.Class.forName0(Native Method)
        at java.lang.Class.forName(Class.java:270)
        at org.apache.spark.serializer.JavaDeserializationStream$$anon$1.resolveClass(JavaSerializer.scala:68)
        at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1612)
        at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1517)
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1771)
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
        at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1990)
        at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1915)
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
        at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1990)
        at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1915)
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
        at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1990)
        at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1915)
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
        at java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
        at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:76)
        at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:115)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)
        at org.apache.spark.scheduler.Task.run(Task.scala:89)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
        at java.lang.Thread.run(Thread.java:745)

resultFlag = Error, returnObj = None
4
  • Apache Zeppelin doing extractly this, writing code on a web page, and execute it. Perhaps you can have a look to Zeppelin and see how it solves the issue. Jun 30, 2016 at 5:34
  • Hi Rockie, really appreciate for your information. I will look the Apache Zeppelin. Jun 30, 2016 at 5:58
  • Hi, I am having similar requirement to be implemented. Have you solved it? May i know which scala and Spark versions you used, since i cannot resolve dependencies when i tried to do the same.
    – NehaM
    Jul 18, 2016 at 14:23
  • Spark Shell handle it by ExecutorClassLoader.scala Aug 23, 2019 at 18:37

1 Answer 1

1

Checked with Apache Zeppelin implementation. Here is the code snippet which used to interpret the input.

It basic uses org.apache.spark.repl.SparkILoop with some code take care of dependency.

public InterpreterResult interpretInput(String[] lines, InterpreterContext context) {
  SparkEnv.set(env);

  // 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(\"\")";

  Console.setOut(context.out);
  out.setInterpreterOutput(context.out);
  context.out.clear();
  Code r = null;
  String incomplete = "";
  boolean inComment = false;

  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();
      boolean continuation = false;
      if (nextLine.isEmpty()
         || nextLine.startsWith("//")         // skip empty line or comment
         || nextLine.startsWith("}")
         || nextLine.startsWith("object")) {  // include "} object" for Scala companion object
        continuation = true;
      } else if (!inComment && nextLine.startsWith("/*")) {
        inComment = true;
        continuation = true;
      } else if (inComment && nextLine.lastIndexOf("*/") >= 0) {
        inComment = false;
        continuation = true;
      } else if (nextLine.length() > 1
              && nextLine.charAt(0) == '.'
              && nextLine.charAt(1) != '.'     // ".."
              && nextLine.charAt(1) != '/') {  // "./"
        continuation = true;
      } else if (inComment) {
        continuation = true;
      }
      if (continuation) {
        incomplete += s + "\n";
        continue;
      }
    }

    scala.tools.nsc.interpreter.Results.Result res = null;
    try {
      res = intp.interpret(incomplete + s);
    } catch (Exception e) {
      sc.clearJobGroup();
      out.setInterpreterOutput(null);
      logger.info("Interpreter exception", e);
      return new InterpreterResult(Code.ERROR, InterpreterUtils.getMostRelevantMessage(e));
    }

    r = getResultCode(res);

    if (r == Code.ERROR) {
      sc.clearJobGroup();
      out.setInterpreterOutput(null);
      return new InterpreterResult(r, "");
    } else if (r == Code.INCOMPLETE) {
      incomplete += s + "\n";
    } else {
      incomplete = "";
    }
  }

  if (r == Code.INCOMPLETE) {
    sc.clearJobGroup();
    out.setInterpreterOutput(null);
    return new InterpreterResult(r, "Incomplete expression");
  } else {
    sc.clearJobGroup();
    out.setInterpreterOutput(null);
    return new InterpreterResult(Code.SUCCESS);
  }
}
1
  • Yeah, got it. I see the source code you post, it executes the code piece by "res = intp.interpret(incomplete + s);". Jun 30, 2016 at 8:29

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.