如何确保 flink 作业已完成执行，然后执行一些任务答案

【问题标题】：how to make sure that flink job has finished executing and then perform some tasks如何确保 flink 作业已完成执行，然后执行一些任务
【发布时间】：2017-11-11 17:15:05
【问题描述】：

我想在 flink 作业完成后执行一些任务，我在 Intellij 中运行代码时没有任何问题，但在 shell 文件中运行 Flink jar 时出现问题。我正在使用下面的行来确保 flink 程序的执行完成

//start the execution

JobExecutionResult jobExecutionResult = envrionment.execute(" Started the execution ");

 is_job_finished = jobExecutionResult.isJobExecutionResult();

我不确定，以上检查是否正确？

然后我在下面的方法中使用上面的变量来执行一些任务

    if(print_mode && is_job_finished){



        System.out.println(" \n \n -- System related  variables  -- \n");

        System.out.println(" Stream_join Window length = " + WindowLength_join__ms + " milliseconds");
        System.out.println(" Input rate for stream RR  = " + input_rate_rr_S + " events/second");
        System.out.println("Stream RR Runtime = " + Stream_RR_RunTime_S + " seconds");
        System.out.println(" # raw events in stream RR  = " + Total_Number_Of_Events_in_RR + "\n");

}

有什么建议吗？

【问题讨论】：

嗨，你能知道怎么做吗？

标签： apache-flink flink-streaming flink-cep

【解决方案1】：

您可以将作业侦听器注册到执行环境。

例如

env.registerJobListener(new JobListener {
      //Callback on job submission.
      override def onJobSubmitted(jobClient: JobClient, throwable: Throwable): Unit = {
        if (throwable == null) {
          log.info("SUBMIT SUCCESS")
        } else {
          log.info("FAIL")
        }
      }
    //Callback on job execution finished, successfully or unsuccessfully.
      override def onJobExecuted(jobExecutionResult: JobExecutionResult, throwable: Throwable): Unit = {

        if (throwable == null) {
          log.info("SUCCESS")
        } else {
          log.info("FAIL")
        }
      }
    })

【讨论】：

【解决方案2】：

向您的 StreamExecutionEnvironment 注册 JobListener。

【讨论】：

你能举个例子吗？
mstzn 在 scala 中包含一个示例。你可以在github.com/apache/flink/blob/release-1.11/flink-tests/src/test/…查看更多信息
我的评论是为了让您的回答更有帮助，对于提出问题的人来说，仅仅指向文档可能不足以解决他们的问题。

【解决方案3】：

如果不是 SQL API，JobListener 就是 grate 程序。

如果使用 SQL API，onJobExecuted 将永远不会被调用。我有一个想法，你可以参考一下。 source 是 Kafka，sink 可以使用任何类型。

让我解释一下：

EndSign：跟随最后一个数据。当您的 Flink 作业消耗它时，这意味着分区元素其余部分为空。

关闭逻辑：

当您 flink 作业处理 EndSign 时。 job需要调用JobController，然后JobController计数器+1
直到 JobController 计数器等于分区计数。然后 JobController 会检查消费者组延迟，确保 Flink 作业获取所有数据。
现在，我们知道工作已经完成了

【讨论】：