[ https://issues.apache.org/jira/browse/SPARK-15582?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15308036#comment-15308036 ]
Catalin Alexandru Zamfir commented on SPARK-15582: -------------------------------------------------- So here's the work-around, wished there was a way not to do it this way. In short, we are: - taking the script by text and using CompilationUnit (from Groovy) to compile the script to bytecode; - for each class we append it to the JAR; - the resulting JAR is added to the Spark context using addJar; - all closures must by dehydrated () which comes with its limitations; {noformat} // Compile it Long dateOfNow = DateTime.now ().getMillis (); String nameOfScript = String.format ("ScriptOf%d", dateOfNow); String pathOfJar = String.format ("distributed/fs/path/to/your/groovy-scripts/%s.jar", String.format ("JarOf%d", dateOfNow)); File archiveFile = new File (pathOfJar); Files.createParentDirs (archiveFile); // With resources List<?> compilationList = compileGroovyScript (nameOfScript, sourceCode); try (JarArchiveOutputStream oneJar = new JarArchiveOutputStream (new FileOutputStream (new File (pathOfJar)))) { // For for (Object compileClass : compilationList) { // Append GroovyClass groovyClass = (GroovyClass) compileClass; JarArchiveEntry oneJarEntry = new JarArchiveEntry (String.format ("%s.class", groovyClass.getName ())); oneJarEntry.setSize (groovyClass.getBytes ().length); byte[] bytecodeOfClass = groovyClass.getBytes (); oneJar.putArchiveEntry (oneJarEntry); oneJar.write (bytecodeOfClass); oneJar.closeArchiveEntry (); } // End it up oneJar.finish (); oneJar.close (); } catch (Exception e) { // Do something } // Append the JAR to the execution environment sparkService.getSparkContext ().addJar (pathOfJar); // GroovyShell.evaluate (scriptText, nameOfScript) below; {noformat} Any idea on how this can be improved? (eg. not using the addJar method and the requirement to not dehydrate the Groovy closures?) > Support for Groovy closures > --------------------------- > > Key: SPARK-15582 > URL: https://issues.apache.org/jira/browse/SPARK-15582 > Project: Spark > Issue Type: Improvement > Components: Input/Output, Java API > Affects Versions: 1.6.1, 1.6.2, 2.0.0 > Environment: 6 node Debian 8 based Spark cluster > Reporter: Catalin Alexandru Zamfir > > After fixing SPARK-13599 and running one of our jobs against this fix for > Groovy dependencies (which indeed it fixed), we see the Spark executors stuck > at a ClassNotFound exception when running as a Script (via > GroovyShell.evalute (scriptText)). It seems Spark cannot de-serialize the > closure, or the closure is not received by the executor. > {noformat} > sparkContext.binaryFiles (ourPath).flatMap ({ onePathEntry -> code-block } as > FlatMapFunction).count (); > { onePathEntry -> code-block } denotes a Groovy closure. > {noformat} > There is a groovy-spark example @ > https://github.com/bunions1/groovy-spark-example ... However the above uses a > modified Groovy. If my understanding is correct, Groovy compiles to native > byte-code, which should be easy for Spark to pick-up and use closures. > The above example code fails with this stack-trace: > {noformat} > Caused by: java.lang.ClassNotFoundException: Script1$_run_closure1 > at java.net.URLClassLoader.findClass(URLClassLoader.java:381) > at java.lang.ClassLoader.loadClass(ClassLoader.java:424) > at java.lang.ClassLoader.loadClass(ClassLoader.java:357) > at java.lang.Class.forName0(Native Method) > at java.lang.Class.forName(Class.java:348) > at > org.apache.spark.serializer.JavaDeserializationStream$$anon$1.resolveClass(JavaSerializer.scala:68) > at > java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1613) > at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1518) > at > java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1774) > at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) > at > java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) > at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) > at > java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801) > at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) > at > java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) > at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) > at > java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801) > at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) > at > java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) > at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) > at > java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801) > at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) > at > java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) > at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) > at > java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801) > at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) > at > java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) > at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) > at > java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801) > at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) > at > java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) > at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) > at > java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801) > at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) > at java.io.ObjectInputStream.readObject(ObjectInputStream.java:371) > 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 > {noformat} > Any ideas on how to tackle this, welcomed. I've tried Googling around for > similar issues, but nobody has found a solution. > At least, point me on where to "hack" to make Spark support closures and I'd > share some of my time to make it work. There is SPARK-2171 arguing that > support for this is out of the box, but for projects of a relative complex > size where the driver application is contained/part-of a bigger application > and running on a cluster, things do not seem to work. I don't know if > SPARK-2171 has tried to run outside of a local[] cluster set-up where such > issues can arise. > I saw a couple of people trying to make it to work, but again, they look to > work-arounds (eg. distribution of byte-code manually before needed, adding a > JAR with addJar and other work-arounds). > Can this be done? Where can we look? -- This message was sent by Atlassian JIRA (v6.3.4#6332) --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org