Hi!
I am trying to implement a scala shell for flink.
I've started with a simple scala object who's main function will drop the
user to the interactive scala shell (repl) at one point:
import scala.tools.nsc.interpreter.ILoop
import scala.tools.nsc.Settings
object Job {
def main(args: Array[String]) {
val repl = new ILoop()
repl.settings = new Settings()
// enable this line to use scala in intellij
repl.settings.usejavacp.value = true
repl.createInterpreter()
// start scala interpreter shell
repl.process(repl.settings)
repl.closeInterpreter()
}
}
Now I am trying to execute the word count example as in:
scala> import org.apache.flink.api.scala._
scala> val env = ExecutionEnvironment.getExecutionEnvironment
scala> val text = env.fromElements("To be, or not to be,--that is the
question:--","Whether 'tis nobler in the mind to suffer", "The slings and
arrows of outrageous fortune","Or to take arms against a sea of troubles,")
scala> val counts = text.flatMap { _.toLowerCase.split("\\W+") }.map { (_,
1) }.groupBy(0).sum(1)
scala> counts.print()
scala> env.execute("Flink Scala Api Skeleton")
However I am running into following error:
env.execute("Flink Scala Api Skeleton")
org.apache.flink.runtime.client.JobExecutionException:
java.lang.RuntimeException: The initialization of the DataSource's outputs
caused an error: The type serializer factory could not load its parameters
from the configuration due to missing classes.
at
org.apache.flink.runtime.operators.DataSourceTask.registerInputOutput(DataSourceTask.java:89)
at
org.apache.flink.runtime.execution.RuntimeEnvironment.<init>(RuntimeEnvironment.java:187)
at
org.apache.flink.runtime.taskmanager.TaskManager.submitTask(TaskManager.java:612)
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.flink.runtime.ipc.RPC$Server.call(RPC.java:420)
at org.apache.flink.runtime.ipc.Server$Handler.run(Server.java:949)
Caused by: java.lang.RuntimeException: The type serializer factory could
not load its parameters from the configuration due to missing classes.
at
org.apache.flink.runtime.operators.util.TaskConfig.getTypeSerializerFactory(TaskConfig.java:1086)
at
org.apache.flink.runtime.operators.util.TaskConfig.getOutputSerializer(TaskConfig.java:542)
at
org.apache.flink.runtime.operators.RegularPactTask.getOutputCollector(RegularPactTask.java:1251)
at
org.apache.flink.runtime.operators.RegularPactTask.initOutputs(RegularPactTask.java:1359)
at
org.apache.flink.runtime.operators.DataSourceTask.initOutputs(DataSourceTask.java:288)
at
org.apache.flink.runtime.operators.DataSourceTask.registerInputOutput(DataSourceTask.java:87)
... 8 more
Caused by: java.lang.ClassNotFoundException: $anon$2$$anon$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:274)
at
org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:54)
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.readObject(ObjectInputStream.java:370)
at
org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:274)
at
org.apache.flink.util.InstantiationUtil.readObjectFromConfig(InstantiationUtil.java:236)
at
org.apache.flink.api.java.typeutils.runtime.RuntimeSerializerFactory.readParametersFromConfig(RuntimeSerializerFactory.java:76)
at
org.apache.flink.runtime.operators.util.TaskConfig.getTypeSerializerFactory(TaskConfig.java:1084)
... 13 more
at
org.apache.flink.runtime.client.JobClient.submitJobAndWait(JobClient.java:349)
at org.apache.flink.client.LocalExecutor.executePlan(LocalExecutor.java:239)
at
org.apache.flink.api.java.LocalEnvironment.execute(LocalEnvironment.java:51)
at
org.apache.flink.api.scala.ExecutionEnvironment.execute(ExecutionEnvironment.scala:501)
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 scala.tools.nsc.interpreter.ILoop.reallyInterpret$1(ILoop.scala:760)
at scala.tools.nsc.interpreter.ILoop.interpretStartingWith(ILoop.scala:805)
at scala.tools.nsc.interpreter.ILoop.command(ILoop.scala:717)
at scala.tools.nsc.interpreter.ILoop.processLine$1(ILoop.scala:581)
at scala.tools.nsc.interpreter.ILoop.innerLoop$1(ILoop.scala:588)
at scala.tools.nsc.interpreter.ILoop.loop(ILoop.scala:591)
at
scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply$mcZ$sp(ILoop.scala:882)
at
scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply(ILoop.scala:837)
at
scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply(ILoop.scala:837)
at
scala.tools.nsc.util.ScalaClassLoader$.savingContextLoader(ScalaClassLoader.scala:135)
at scala.tools.nsc.interpreter.ILoop.process(ILoop.scala:837)
at org.myorg.quickstart.Job$.main(Job.scala:37)
at org.myorg.quickstart.Job.main(Job.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 com.intellij.rt.execution.application.AppMain.main(AppMain.java:134)
I'm pretty new to Scala and Flink, so maybe someone has a suggestion or can
point me in some direction?
thanks,
Nikolaas
