从 Java 应用程序启动和监控 Spark 应用程序答案

【问题标题】：Starting and monitoring spark application from a Java application从 Java 应用程序启动和监控 Spark 应用程序
【发布时间】：2017-06-07 06:07:20
【问题描述】：

我正在开发一个应用程序以从 java 应用程序执行 spark 批处理应用程序。

有一个主类启动线程来启动spark 应用程序。它使用zookeeper 在将启动spark 应用程序的机器中查找leader。 Main 方法如下所示：

public static void main(String[] args) throws IOException {

        final int id = Integer.valueOf(args[0]);
        final String zkURL = args[1];

        final ExecutorService service = Executors.newSingleThreadExecutor();

        final Future<?> status = service.submit(new ProcessNode(id, zkURL));
        try {
            status.get();
        } catch (InterruptedException | ExecutionException e) {
            LOG.fatal(e.getMessage(), e);
            service.shutdown();
        }

一旦leader被选中，下面的代码就会在它上面运行以启动spark应用程序。

protected Boolean executeCommand() {
    try {
        final Runtime rt = Runtime.getRuntime();
        final Process proc = rt.exec("sh start-sparkapp.sh");
        final int exitVal = proc.waitFor();
        BufferedReader buf = new BufferedReader(new InputStreamReader(proc.getInputStream()));
        String line = "";
        while ((line=buf.readLine())!=null) {
        System.out.println(line);
        }

        System.out.println(" commandToExecute exited with code: " + exitVal);
        proc.destroy();

    } catch (final Exception e) {
        System.out.println("Exception occurred while Launching process : " + e.getMessage());
        return Boolean.FALSE;
    }
         return Boolean.TRUE;
}

但这会启动一个长期运行的spark 工作。所以我相信，只有在spark 作业完成时才会执行下一部分代码。我的要求是在spark 应用程序启动后，控制转到代码的下一部分，我正在监视相同spark 应用程序的状态。即我启动spark 应用程序并从同一个java 应用程序监视spark 应用程序的状态。假设我有一个方法 montior 来监控应用程序的状态

public String monitor(ApplicationId id)

有什么建议可以实现吗？

【问题讨论】：

标签“apache-spark”是否合理？

标签： java multithreading apache-spark executorservice

【解决方案1】：

由于您将使用 public String monitor(ApplicationId id) 方法监视 Spark 应用程序，我假设您不希望当前线程使用 proc.waitFor() 等待进程。此外，您不想将进程的正常输出打印到控制台。这两个操作都使您的线程等待生成的进程。此外，您的监控方法应该将生成的进程的进程 ID，而不是 spark applicationId 作为输入。所以修改后的代码可能如下所示：

protected Boolean executeCommand() {
try {
    final Runtime rt = Runtime.getRuntime();
    final Process proc = rt.exec("sh start-sparkapp.sh");

    /* 
     *Call to method monitor(ProcessId id)
     */

    } catch (final Exception e) {
        System.out.println("Exception occurred while Launching process : " + e.getMessage());
        return Boolean.FALSE;
    }
     return Boolean.TRUE;
}

【讨论】：