从Flink作业UI提交作业的问题(异常:org.apache.flink.client.program.OptimizerPlanEnvironment $ ProgramAbortExceptio

本文介绍了从Flink作业UI提交作业的问题(异常:org.apache.flink.client.program.OptimizerPlanEnvironment $ ProgramAbortException)的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

我有用于flink作业的简单Java代码

I have simple java code for flink job

List<Tuple2> list = new ArrayList<>(); for (int i = 0; i < 10; i++) { list.add(new Tuple2(Integer.valueOf(i), "test" + i)); } StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.fromCollection(list).print(); env.execute("job1");

我打包了这段代码并创建了一个jar:说flink-processor-0.1-SNAPSHOT.jar，然后将其从Submit job UI上载到JobManager.上传没有问题.我看到EntryClass具有主类(com.abc.xyz.streaming.FlinkProcessor)现在，我使用一些参数(--ns.conf1 abc.file --ns.conf2 xyz.file)从"Submit Job Ui"提交作业，并指定了主类(com.abc.xyz.streaming.FlinkProcessor).作业提交失败..在JobManager中，我看到以下错误.

I packaged this code and create a jar: say flink-processor-0.1-SNAPSHOT.jar, upload it to JobManager from Submit job UI. No issues in upload. I see the EntryClass has the main class (com.abc.xyz.streaming.FlinkProcessor) Now, i submit the job from "Submit Job Ui "with some params (--ns.conf1 abc.file --ns.conf2 xyz.file) and specified the main class (com.abc.xyz.streaming.FlinkProcessor). The job submission fails. In the JobManager, i see the following errors.

org.apache.flink.client.program.OptimizerPlanEnvironment$ProgramAbortException at org.apache.flink.streaming.api.environment.StreamPlanEnvironment.execute(StreamPlanEnvironment.java:70) at org.apache.flink.streaming.api.environment.StreamPlanEnvironment.execute(StreamPlanEnvironment.java:53) at com.abc.xyz.streaming.FlinkProcessor.run(FlinkProcessor.java:114) at com.abc.xyz.streaming.FlinkProcessor.main(FlinkProcessor.java:53) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessor at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:529) at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:421) at org.apache.flink.client.program.OptimizerPlanEnvironment.getOptimizedPlan(OptimizerPlanEnvironment.java:83) at org.apache.flink.client.program.PackagedProgramUtils.createJobGraph(PackagedProgramUtils.java:80) at org.apache.flink.runtime.webmonitor.handlers.utils.JarHandlerUtils$JarHandlerContext.toJobGraph(JarHandlerUtils.java:126) at org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$getJobGraphAsync$6(JarRunHandler.java:142) at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) 2020-06-16 16:14:20,588 ERROR org.apache.flink.runtime.webmonitor.handlers.JarRunHandler - Unhandled exception. org.apache.flink.client.program.ProgramInvocationException: The program plan could not be fetched - the program aborted pre-maturely. System.err: Running flink job :2020-06-16T23:14:20.476Z Error msg: org.apache.flink.client.program.OptimizerPlanEnvironment$ProgramAbortException at org.apache.flink.streaming.api.environment.StreamPlanEnvironment.execute(StreamPlanEnvironment.java:70) at org.apache.flink.streaming.api.environment.StreamPlanEnvironment.execute(StreamPlanEnvironment.java:53)

当我使用flink run命令提交同一作业时，它工作正常.没有错误.

flink运行-c com.abc.xyz.streaming.FlinkProcessor/Users/abc/target/flink-processor-0.1-SNAPSHOT.jar --ns.conf1 abc.file --ns.conf2 xyz.文件

不确定我在这里想念什么吗?非常感谢您的帮助.

Not sure what I am missing here? Any help is much appreciated.

推荐答案

发现了问题.我用

try { List<Tuple2> list = new ArrayList<>(); for (int i = 0; i < 10; i++) { list.add(new Tuple2(Integer.valueOf(i), "test" + i)); } StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.fromCollection(list).print(); env.execute("job1"); } catch (Throwable t) { ignore t };

更改了异常块尝试{.....} catch(Exception exp){ignore exp}; 此后，它开始工作.

changed the exception block try { ..... } catch (Exception exp) { ignore exp}; After this, it started working.

谢谢！