Когда я создаю проект с использованием шаблона sbt, рекомендованного: https://ci.apache.org/projects/flink/flink-docs-release-1.6/quickstart/scala_api_quickstart.html с
sbt new tillrohrmann/flink-project.g8
Я получаю сообщение об ошибке, показанное ниже, когда пытаюсь предоставить любую анонимную функцию для методов scala API(фильтр, карта и т. д.).ordinals.print () работает нормально.Я не в реплинге scala, это просто нормальный реплик в scala.Согласно https://ci.apache.org/projects/flink/flink-docs-stable/dev/local_execution.html#collection-environment все должно быть в порядке, и оно должно работать, если я собираюсь использовать это (и StreamExecutionEnvironment) в интеграционных тестах.Я знаю о CollectionEnvironment, и он отлично работает, но это не помогает мне с StreamExecutionEnvironment.Я пробовал это в 2 средах intellij, в windows и linux, оба с использованием scala 2.11.12, flink 1.6.1 и java 8.
Конечно, документы среды исполнения основаны на java--Есть ли какая-то зависимость или импорт, которые мне нужны, чтобы заставить работать API Scala?
scala> import org.apache.flink.api.scala.ExecutionEnvironment
val benv = ExecutionEnvironment.getExecutionEnvironment
import org.apache.flink.api.scala.ExecutionEnvironment
scala> SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
benv: org.apache.flink.api.scala.ExecutionEnvironment = org.apache.flink.api.scala.ExecutionEnvironment@646427f7
scala> import org.apache.flink.api.scala._
import org.apache.flink.api.scala._
scala> val ordinals = benv.fromCollection(65 until 65 + 26)
ordinals: org.apache.flink.api.scala.DataSet[Int] = org.apache.flink.api.scala.DataSet@40017e98
scala> val capitals = ordinals.map(_.toChar)
capitals: org.apache.flink.api.scala.DataSet[Char] = org.apache.flink.api.scala.DataSet@7323dce7
scala> capitals.print()
org.apache.flink.runtime.client.JobExecutionException: java.lang.RuntimeException: The initialization of the DataSource's outputs caused an error: Could not read the user code wrapper: $anonfun$1
at org.apache.flink.runtime.minicluster.MiniCluster.executeJobBlocking(MiniCluster.java:623)
at org.apache.flink.client.LocalExecutor.executePlan(LocalExecutor.java:235)
at org.apache.flink.api.java.LocalEnvironment.execute(LocalEnvironment.java:91)
at org.apache.flink.api.java.ExecutionEnvironment.execute(ExecutionEnvironment.java:816)
at org.apache.flink.api.java.DataSet.collect(DataSet.java:413)
at org.apache.flink.api.java.DataSet.print(DataSet.java:1652)
at org.apache.flink.api.scala.DataSet.print(DataSet.scala:1726)
... 28 elided
Caused by: java.lang.RuntimeException: The initialization of the DataSource's outputs caused an error: Could not read the user code wrapper: $anonfun$1
at org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:106)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.runtime.operators.util.CorruptConfigurationException: Could not read the user code wrapper: $anonfun$1
at org.apache.flink.runtime.operators.util.TaskConfig.getStubWrapper(TaskConfig.java:290)
at org.apache.flink.runtime.operators.BatchTask.instantiateUserCode(BatchTask.java:1445)
at org.apache.flink.runtime.operators.chaining.ChainedMapDriver.setup(ChainedMapDriver.java:39)
at org.apache.flink.runtime.operators.chaining.ChainedDriver.setup(ChainedDriver.java:90)
at org.apache.flink.runtime.operators.BatchTask.initOutputs(BatchTask.java:1312)
at org.apache.flink.runtime.operators.DataSourceTask.initOutputs(DataSourceTask.java:312)
at org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:103)
... 2 more
Caused by: java.lang.ClassNotFoundException: $anonfun$1
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$ChildFirstClassLoader.loadClass(FlinkUserCodeClassLoaders.java:129)
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.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:76)
at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1866)
at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1749)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2040)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1571)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2285)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2209)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2067)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1571)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2285)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2209)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2067)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1571)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:431)
at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:502)
at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:489)
at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:477)
at org.apache.flink.util.InstantiationUtil.readObjectFromConfig(InstantiationUtil.java:438)
at org.apache.flink.runtime.operators.util.TaskConfig.getStubWrapper(TaskConfig.java:288)
... 8 more
scala>
update: изнутри скалярного, ExecutionEnvironment.getExecutionEnvironment и StreamExecutionEnvironment.getExecutionEnvironment работают без ошибок, поэтому я гораздо меньше обеспокоен.