Javaアプリケーションからのスパークアプリケーションの起動と監視

Question

javaアプリケーションからsparkバッチアプリケーションを実行するアプリケーションで作業しています。

sparkアプリケーションを開始するスレッドを開始するメインクラスが1つあります。これはzookeeperを使用してsparkアプリケーションを開始するマシンの中でleaderを検索します。 leaderを選択するとコードがsparkアプリケーションを起動することで実行します以下、

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(); 
     } 

：Main方法は次のようになります。

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アプリケーションのステータスを監視します。 私はどのようにこれを達成するために、アプリケーション

public String monitor(ApplicationId id) 

どれ提案の状態を監視する方法montiorを持っていると仮定しますか？

Answer 1

public String monitor(ApplicationId id)という方法でSparkアプリケーションを監視しているので、私はあなたの現在のスレッドがproc.waitFor()を使ってプロセスを待つことを望まないと仮定しています。さらに、プロセスの通常の出力をコンソールに出力したくない場合もあります。どちらの操作でも、スレッドは生成されたプロセスを待機します。さらに、モニターメソッドは、入力としてspark applicationIdではなく、生成されたプロセスのprocess-idを取る必要があります。 変更されたコードは次のようになります。

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; 
}