如何使用 Java 代码将作业提交到 Flink 集群?

Question

我已经将一个包含我的应用程序代码的胖罐上传到我的 Flink 集群中所有节点的/lib 文件夹中。我正在尝试从单独的 Java 应用程序启动 Flink 作业，但找不到这样做的好方法。

我目前找到的最接近解决方案的是具有运行作业 API 的监控 Rest API。但是，这仅允许您运行通过作业上传功能提交的作业。

我已经在 flink-client 模块中看到了ClusterClient.java，但看不到任何如何使用它的示例。

任何关于某人如何通过java代码成功提交作业的例子将不胜感激！

Answer 1

您可以使用RestClusterClient运行指向您的 Flink 作业的PackagedProgram。如果你的作业接受一些参数，你可以传递它们。

下面是在localhost:8081上运行的独立集群的示例：

// import org.apache.flink.api.common.JobSubmissionResult;
// import org.apache.flink.client.deployment.StandaloneClusterId;
// import org.apache.flink.client.program.PackagedProgram;
// import org.apache.flink.client.program.rest.RestClusterClient;
// import org.apache.flink.configuration.Configuration;
// import org.apache.flink.configuration.JobManagerOptions;
// import org.apache.flink.configuration.RestOptions;
String clusterHost = "localhost";
int clusterPort = 8081;
Configuration config = new Configuration();
config.setString(JobManagerOptions.ADDRESS, clusterHost);
config.setInteger(RestOptions.PORT, clusterPort);
String jarFilePath = "/opt/flink/examples/streaming/SocketWindowWordCount.jar";
String[] args = new String[]{ "--port", "9000" };
PackagedProgram packagedProgram = new PackagedProgram(new File(jarFilePath), args);
RestClusterClient<StandaloneClusterId> client =
new RestClusterClient<StandaloneClusterId>(config, StandaloneClusterId.getInstance());
int parallelism = 1;
JobSubmissionResult result = client.run(packagedProgram,  parallelism);

Answer 2

这似乎适用于1.10版

private static final int PARALLELISM = 8;
private static final Configuration FLINK_CONFIG = new Configuration();
void foo() throws Exception {
FLINK_CONFIG.setString(JobManagerOptions.ADDRESS, "localhost");
FLINK_CONFIG.setInteger(RestOptions.PORT, 8081);
FLINK_CONFIG.setInteger(RestOptions.RETRY_MAX_ATTEMPTS, 3);
RestClusterClient<StandaloneClusterId> flinkClient = new RestClusterClient<>(FLINK_CONFIG, StandaloneClusterId.getInstance());
String jar = "/path/to/jar";
String[] args = new String[]{"..."};
PackagedProgram program = PackagedProgram.newBuilder()
.setJarFile(new File(jar))
.setArguments(args)
.build();
JobGraph jobGraph = PackagedProgramUtils.createJobGraph(program, FLINK_CONFIG, PARALLELISM, false);
JobID jobId = flinkClient.submitJob(jobGraph).get();
...
}