我正在使用ApacheFlink-1.1.3进行实时流数据分析。我的系统由用于消息队列的Kafka集群、从Kafka分区读取消息并对其进行分析的Flink集群组成,最后我希望将生成的数据转储到Ignite缓存中。对于系统,我使用IgniteSink类将数据放入ignite缓存。版本如下:
Flink1.1.3,
Kafka2.10,点燃2.0.0
当我试图在flink cluster上运行作业时,它给出了以下错误:,
Exception in thread "main" org.apache.flink.client.program.ProgramInvocationException: The program execution failed: Job execution failed.
at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:409)
at org.apache.flink.client.program.StandaloneClusterClient.submitJob(StandaloneClusterClient.java:95)
at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:382)
at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:374)
at org.apache.flink.streaming.api.environment.RemoteStreamEnvironment.executeRemotely(RemoteStreamEnvironment.java:209)
at org.apache.flink.streaming.api.environment.RemoteStreamEnvironment.execute(RemoteStreamEnvironment.java:173)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1429)
at flink_ignite_sink_remote.main(flink_ignite_sink_remote.java:77)
Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$8.apply$mcV$sp(JobManager.scala:822)
at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$8.apply(JobManager.scala:768)
at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$8.apply(JobManager.scala:768)
at scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)
at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:41)
at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:401)
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.pollAndExecAll(ForkJoinPool.java:1253)
at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1346)
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
Caused by: java.lang.NoClassDefFoundError: Could not initialize class org.apache.ignite.sink.flink.IgniteSink$SinkContext$Holder
at org.apache.ignite.sink.flink.IgniteSink$SinkContext.getStreamer(IgniteSink.java:201)
at org.apache.ignite.sink.flink.IgniteSink$SinkContext.access$100(IgniteSink.java:175)
at org.apache.ignite.sink.flink.IgniteSink.invoke(IgniteSink.java:165)
at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:39)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:373)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:358)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:346)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:329)
at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:51)
at flink_ignite_sink_remote$Splitter.flatMap(flink_ignite_sink_remote.java:177)
at flink_ignite_sink_remote$Splitter.flatMap(flink_ignite_sink_remote.java:1)
at org.apache.flink.streaming.api.operators.StreamFlatMap.processElement(StreamFlatMap.java:48)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:373)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:358)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:346)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:329)
at org.apache.flink.streaming.api.operators.StreamSource$NonTimestampContext.collect(StreamSource.java:161)
at org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecord(AbstractFetcher.java:225)
at org.apache.flink.streaming.connectors.kafka.internal.Kafka09Fetcher.run(Kafka09Fetcher.java:253)
at java.lang.Thread.run(Thread.java:745)
我已经将Ignite Flink的所有库都包含在我的项目中,但我仍然得到了java。lang.NoClassDefFoundError
错误。
使用部署在flink群集上的flink作业时,有两个选项:
您的问题似乎不是生成一个包含所有依赖项的jar文件,而且这些依赖项不在fink服务器的类路径中。
尝试运行以下mvn命令以生成JAR:
mvn清洁包-Pbuild jar
它可以生成多个罐子,选择胖罐子(越大)
Apache Ignite用户论坛也讨论了这一点:http://apache-ignite-users.70518.x6.nabble.com/Flink-Streamer-td10650.html
我怀疑你用的是一个简单的罐子,而不是超级/胖罐子。
如果您正在使用maven,请尝试shade插件或用于sbt sbt组装。您还可以按照《快速入门指南》中的说明创建项目
kubectl get Pods-N队列-前门 客户端XML SpringConfig文件
我使用子JPanel构造JScrollPane,然后尝试将JScrollPane添加到父JPanel中,希望有一个可滚动的自定义JPanel。 我有一个大的细白线前面的所有我的组件,他们是没有滚动。有什么想法吗?
我已经为android Studio创建了一个webview应用程序。但没有加载web URL。错误为NET::ERR_ACCESS_DENIED。有谁能帮忙吗
我有一个使用Java Version8的Spring和Maven的项目。当我运行maven update时,它适用于Java9或更高版本,当我编译它时,它适用于Java8。我注意到了这一点,因为用Maven更新会将一个特定的类更改为Java9,而我无法导入这个类。 这个类是:javax.annotation.generated(Java8)。 这是要导入的类。但是,当我进行maven更新时,文件的
当应用程序启动EncryptionBootstrapConfiguration无法自动装配我的自定义TextEncryptor-https://github.com/spring-cloud/spring-cloud-commons/blob/cde7c7f3118382490c28776f66e0a56f248141fd/spring-cloud-context/src/main/java/or
我有一个问题,找出我的源代码中的错误。 有什么办法可以解决它吗? 我的代码中抛出了“ArrayIndexOutOfBoundsException”,但我无法找到它的位置。 android studio中的logcat:
RISC-V 与中断相关的寄存器和指令 [info] 回顾:RISC-V 中的机器态(Machine Mode,机器模式,M 模式) 是 RISC-V 中的最高权限模式,一些底层操作的指令只能由机器态进行使用。 是所有标准 RISC-V 处理器都必须实现的模式。 默认所有中断实际上是交给机器态处理的,但是为了实现更多功能,机器态会将某些中断交由内核态处理。这些异常也正是我们编写操作系统所需要实现的
加载EGL入口点时出错。 Java运行时环境检测到一个致命错误: exception_access_visultion(0xC0000005)在pc=0x000000000000,pid=22096,tid=21896 这是错误日志文件中的线程(完整文件太长):------------------------------------------------- 非常感谢你的回答。埃吉尔