Re: FlinkJobNotFoundException
Hi Doug, sorry for being not responsive the last two weeks. Other stuff kept me busy. A few things to note on your issue: It looks like the job result is requested while doing the job execution in a synchronous way. Flink will try to access the ArchivedExecutionGraphStore to get the job's result after it finishes. By default, Flink's session mode relies on the FileArchivedExecutionGraphStore which relies on a temporary folder. If this temporary folder gets cleaned up in any way, Flink would not be able to retrieve the job result if the number of jobs being stored in this ArchivedExecutionGraphStore exceeds the size of the internally used cache. Same might be true if the underlying file is corrupted in any way. This brings up the following questions/action items: - Could you provide the debug logs for this? - Could you provide the Flink configuration you use to check the cache configuration? - Why do you commit two jobs with the same job ID? This might cause the job's execution graph to be overwritten in the temporary folder and, therefore, might lead to a corrupted file while writing the second job's ExecutionGraph to disk. As a consequence another REST request that is sent later on to the Flink cluster's REST endpoint with the same job ID should succeed again. That would be something to try out. - Could you monitor Flink's temporary folder and see whether there is some cleanup happening by some other process? Best, Matthias On Wed, Oct 13, 2021 at 3:12 PM Gusick, Doug S wrote: > Hi Matthias, > > > > Do you have any update here? > > > > Thank you, > > Doug > > > > *From:* Gusick, Doug S [Engineering] > *Sent:* Thursday, October 7, 2021 9:03 AM > *To:* Hailu, Andreas [Engineering] ; > Matthias Pohl > *Cc:* user@flink.apache.org; Erai, Rahul [Engineering] < > rahul.e...@ny.email.gs.com> > *Subject:* RE: FlinkJobNotFoundException > > > > Hi Matthias, > > > > I just wanted to follow up here. Were you able to access the jobmanager > log? If so, were you able to find anything around the issues we have been > facing? > > > > Best, > > Doug > > > > *From:* Hailu, Andreas [Engineering] > *Sent:* Thursday, September 30, 2021 8:56 AM > *To:* Matthias Pohl ; Gusick, Doug S > [Engineering] > *Cc:* user@flink.apache.org; Erai, Rahul [Engineering] < > rahul.e...@ny.email.gs.com> > *Subject:* RE: FlinkJobNotFoundException > > > > Hi Matthias, the log file is quite large (21MB) so mailing it over in its > entirety may have been a challenge. The file is available here [1], and > we’re of course happy to share any relevant parts of it with the mailing > list. > > > > I think since we’ve shared logs with you before in the past, you weren’t > sent over an additional welcome email J > > > > [1] > https://lockbox.gs.com/lockbox/folders/dc2ccacc-f2d2-4d66-a098-461b43e8b65f/ > > > > *// *ah > > > > *From:* Matthias Pohl > *Sent:* Thursday, September 30, 2021 2:57 AM > *To:* Gusick, Doug S [Engineering] > *Cc:* user@flink.apache.org; Erai, Rahul [Engineering] < > rahul.e...@ny.email.gs.com> > *Subject:* Re: FlinkJobNotFoundException > > > > I didn't receive any email. But we rather not do individual support. > Please share the logs on the mailing list. This way, anyone is able to > participate in the discussion. > > > > Best, > Matthias > > > > On Wed, Sep 29, 2021 at 8:12 PM Gusick, Doug S wrote: > > Hi Matthias, > > > > Thank you for getting back. We have been looking into upgrading to a newer > version, but have not completed full testing just yet. > > > > I was unable to find a previous error in the JM logs. You should have > received an email with details to a “lockbox”. I have uploaded the job > manager logs there. Please let me know if you need any more information. > > > > Thank you, > > Doug > > > > *From:* Matthias Pohl > *Sent:* Wednesday, September 29, 2021 12:00 PM > *To:* Gusick, Doug S [Engineering] > *Cc:* user@flink.apache.org; Erai, Rahul [Engineering] < > rahul.e...@ny.email.gs.com> > *Subject:* Re: FlinkJobNotFoundException > > > > Hi Doug, > > thanks for reaching out to the community. First of all, 1.9.2 is quite an > old Flink version. You might want to consider upgrading to a newer version. > The community only offers support for the two most-recent Flink versions. > Newer version might include fixes for your issue. > > > > But back to your actual problem: The logs you're providing only show that > some job switched into FINISHED state. Is there some error showing up > earlier in the logs which you might have missed? It would be helpful if you > could share the complet
RE: FlinkJobNotFoundException
Hi Matthias, Do you have any update here? Thank you, Doug From: Gusick, Doug S [Engineering] Sent: Thursday, October 7, 2021 9:03 AM To: Hailu, Andreas [Engineering] ; Matthias Pohl Cc: user@flink.apache.org; Erai, Rahul [Engineering] Subject: RE: FlinkJobNotFoundException Hi Matthias, I just wanted to follow up here. Were you able to access the jobmanager log? If so, were you able to find anything around the issues we have been facing? Best, Doug From: Hailu, Andreas [Engineering] mailto:andreas.ha...@ny.email.gs.com>> Sent: Thursday, September 30, 2021 8:56 AM To: Matthias Pohl mailto:matth...@ververica.com>>; Gusick, Doug S [Engineering] mailto:doug.gus...@ny.email.gs.com>> Cc: user@flink.apache.org<mailto:user@flink.apache.org>; Erai, Rahul [Engineering] mailto:rahul.e...@ny.email.gs.com>> Subject: RE: FlinkJobNotFoundException Hi Matthias, the log file is quite large (21MB) so mailing it over in its entirety may have been a challenge. The file is available here [1], and we’re of course happy to share any relevant parts of it with the mailing list. I think since we’ve shared logs with you before in the past, you weren’t sent over an additional welcome email ☺ [1] https://lockbox.gs.com/lockbox/folders/dc2ccacc-f2d2-4d66-a098-461b43e8b65f/ // ah From: Matthias Pohl mailto:matth...@ververica.com>> Sent: Thursday, September 30, 2021 2:57 AM To: Gusick, Doug S [Engineering] mailto:doug.gus...@ny.email.gs.com>> Cc: user@flink.apache.org<mailto:user@flink.apache.org>; Erai, Rahul [Engineering] mailto:rahul.e...@ny.email.gs.com>> Subject: Re: FlinkJobNotFoundException I didn't receive any email. But we rather not do individual support. Please share the logs on the mailing list. This way, anyone is able to participate in the discussion. Best, Matthias On Wed, Sep 29, 2021 at 8:12 PM Gusick, Doug S mailto:doug.gus...@gs.com>> wrote: Hi Matthias, Thank you for getting back. We have been looking into upgrading to a newer version, but have not completed full testing just yet. I was unable to find a previous error in the JM logs. You should have received an email with details to a “lockbox”. I have uploaded the job manager logs there. Please let me know if you need any more information. Thank you, Doug From: Matthias Pohl mailto:matth...@ververica.com>> Sent: Wednesday, September 29, 2021 12:00 PM To: Gusick, Doug S [Engineering] mailto:doug.gus...@ny.email.gs.com>> Cc: user@flink.apache.org<mailto:user@flink.apache.org>; Erai, Rahul [Engineering] mailto:rahul.e...@ny.email.gs.com>> Subject: Re: FlinkJobNotFoundException Hi Doug, thanks for reaching out to the community. First of all, 1.9.2 is quite an old Flink version. You might want to consider upgrading to a newer version. The community only offers support for the two most-recent Flink versions. Newer version might include fixes for your issue. But back to your actual problem: The logs you're providing only show that some job switched into FINISHED state. Is there some error showing up earlier in the logs which you might have missed? It would be helpful if you could share the complete JobManager logs to get a better understanding of what's going on. Best, Matthias On Wed, Sep 29, 2021 at 3:47 PM Gusick, Doug S mailto:doug.gus...@gs.com>> wrote: Hello, We are facing an issue with some of our applications that are submitting a high volume of jobs to Flink (we are using v1.9.2). We are observing that numerous jobs (in this case 44 out of 350+) fail with the same FlinkJobNotFoundException within a 45 second timeframe. From our client logs, this is the exception we can see: Calc Engine: Caused by: org.apache.flink.runtime.rest.util.RestClientException: [org.apache.flink.runtime.messages.FlinkJobNotFoundException: Could not find Flink job (d0991f0ae712a9df710aa03311a32c8c)] Calc Engine: at org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:389) Calc Engine: at org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:373) Calc Engine: at java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:952) Calc Engine: at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926) Calc Engine: at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442) Calc Engine: ... 3 more This is the first job to fail with the above exception. From the JobManager logs, we can see that the job goes to FINISHED State, and then we see the following exception: 2021-09-28 04:54:16,936 INFO [flink-akka.actor.default-dispatcher-28] org.apache.flink.runtime.executiongraph.ExecutionGraph- Job Flink Java Job at Tue Sep 28 04:48:21 EDT 2021 (d0991f0ae712a9df710aa03311a32c8c) switched from state RUNNING to FINISHED. 2021-09-28 04:54:16,937 INFO [flink-akka.actor.default-dispatcher-28] org.apache.
RE: FlinkJobNotFoundException
Hi Matthias, I just wanted to follow up here. Were you able to access the jobmanager log? If so, were you able to find anything around the issues we have been facing? Best, Doug From: Hailu, Andreas [Engineering] Sent: Thursday, September 30, 2021 8:56 AM To: Matthias Pohl ; Gusick, Doug S [Engineering] Cc: user@flink.apache.org; Erai, Rahul [Engineering] Subject: RE: FlinkJobNotFoundException Hi Matthias, the log file is quite large (21MB) so mailing it over in its entirety may have been a challenge. The file is available here [1], and we’re of course happy to share any relevant parts of it with the mailing list. I think since we’ve shared logs with you before in the past, you weren’t sent over an additional welcome email ☺ [1] https://lockbox.gs.com/lockbox/folders/dc2ccacc-f2d2-4d66-a098-461b43e8b65f/ // ah From: Matthias Pohl mailto:matth...@ververica.com>> Sent: Thursday, September 30, 2021 2:57 AM To: Gusick, Doug S [Engineering] mailto:doug.gus...@ny.email.gs.com>> Cc: user@flink.apache.org<mailto:user@flink.apache.org>; Erai, Rahul [Engineering] mailto:rahul.e...@ny.email.gs.com>> Subject: Re: FlinkJobNotFoundException I didn't receive any email. But we rather not do individual support. Please share the logs on the mailing list. This way, anyone is able to participate in the discussion. Best, Matthias On Wed, Sep 29, 2021 at 8:12 PM Gusick, Doug S mailto:doug.gus...@gs.com>> wrote: Hi Matthias, Thank you for getting back. We have been looking into upgrading to a newer version, but have not completed full testing just yet. I was unable to find a previous error in the JM logs. You should have received an email with details to a “lockbox”. I have uploaded the job manager logs there. Please let me know if you need any more information. Thank you, Doug From: Matthias Pohl mailto:matth...@ververica.com>> Sent: Wednesday, September 29, 2021 12:00 PM To: Gusick, Doug S [Engineering] mailto:doug.gus...@ny.email.gs.com>> Cc: user@flink.apache.org<mailto:user@flink.apache.org>; Erai, Rahul [Engineering] mailto:rahul.e...@ny.email.gs.com>> Subject: Re: FlinkJobNotFoundException Hi Doug, thanks for reaching out to the community. First of all, 1.9.2 is quite an old Flink version. You might want to consider upgrading to a newer version. The community only offers support for the two most-recent Flink versions. Newer version might include fixes for your issue. But back to your actual problem: The logs you're providing only show that some job switched into FINISHED state. Is there some error showing up earlier in the logs which you might have missed? It would be helpful if you could share the complete JobManager logs to get a better understanding of what's going on. Best, Matthias On Wed, Sep 29, 2021 at 3:47 PM Gusick, Doug S mailto:doug.gus...@gs.com>> wrote: Hello, We are facing an issue with some of our applications that are submitting a high volume of jobs to Flink (we are using v1.9.2). We are observing that numerous jobs (in this case 44 out of 350+) fail with the same FlinkJobNotFoundException within a 45 second timeframe. From our client logs, this is the exception we can see: Calc Engine: Caused by: org.apache.flink.runtime.rest.util.RestClientException: [org.apache.flink.runtime.messages.FlinkJobNotFoundException: Could not find Flink job (d0991f0ae712a9df710aa03311a32c8c)] Calc Engine: at org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:389) Calc Engine: at org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:373) Calc Engine: at java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:952) Calc Engine: at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926) Calc Engine: at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442) Calc Engine: ... 3 more This is the first job to fail with the above exception. From the JobManager logs, we can see that the job goes to FINISHED State, and then we see the following exception: 2021-09-28 04:54:16,936 INFO [flink-akka.actor.default-dispatcher-28] org.apache.flink.runtime.executiongraph.ExecutionGraph- Job Flink Java Job at Tue Sep 28 04:48:21 EDT 2021 (d0991f0ae712a9df710aa03311a32c8c) switched from state RUNNING to FINISHED. 2021-09-28 04:54:16,937 INFO [flink-akka.actor.default-dispatcher-28] org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Job d0991f0ae712a9df710aa03311a32c8c reached globally terminal state FINISHED. 2021-09-28 04:54:16,939 INFO [flink-akka.actor.default-dispatcher-28] org.apache.flink.runtime.jobmaster.JobMaster - Stopping the JobMaster for job Flink Java Job at Tue Sep 28 04:48:21 EDT 2021(d0991f0ae712a9df710aa03311a32c8c). 2021-09-28 04:54:16,940 INFO [flink-akka.actor.default-dispatcher-39] org.apache.flink.yarn.YarnR
RE: FlinkJobNotFoundException
Hi Matthias, the log file is quite large (21MB) so mailing it over in its entirety may have been a challenge. The file is available here [1], and we’re of course happy to share any relevant parts of it with the mailing list. I think since we’ve shared logs with you before in the past, you weren’t sent over an additional welcome email ☺ [1] https://lockbox.gs.com/lockbox/folders/dc2ccacc-f2d2-4d66-a098-461b43e8b65f/ // ah From: Matthias Pohl Sent: Thursday, September 30, 2021 2:57 AM To: Gusick, Doug S [Engineering] Cc: user@flink.apache.org; Erai, Rahul [Engineering] Subject: Re: FlinkJobNotFoundException I didn't receive any email. But we rather not do individual support. Please share the logs on the mailing list. This way, anyone is able to participate in the discussion. Best, Matthias On Wed, Sep 29, 2021 at 8:12 PM Gusick, Doug S mailto:doug.gus...@gs.com>> wrote: Hi Matthias, Thank you for getting back. We have been looking into upgrading to a newer version, but have not completed full testing just yet. I was unable to find a previous error in the JM logs. You should have received an email with details to a “lockbox”. I have uploaded the job manager logs there. Please let me know if you need any more information. Thank you, Doug From: Matthias Pohl mailto:matth...@ververica.com>> Sent: Wednesday, September 29, 2021 12:00 PM To: Gusick, Doug S [Engineering] mailto:doug.gus...@ny.email.gs.com>> Cc: user@flink.apache.org<mailto:user@flink.apache.org>; Erai, Rahul [Engineering] mailto:rahul.e...@ny.email.gs.com>> Subject: Re: FlinkJobNotFoundException Hi Doug, thanks for reaching out to the community. First of all, 1.9.2 is quite an old Flink version. You might want to consider upgrading to a newer version. The community only offers support for the two most-recent Flink versions. Newer version might include fixes for your issue. But back to your actual problem: The logs you're providing only show that some job switched into FINISHED state. Is there some error showing up earlier in the logs which you might have missed? It would be helpful if you could share the complete JobManager logs to get a better understanding of what's going on. Best, Matthias On Wed, Sep 29, 2021 at 3:47 PM Gusick, Doug S mailto:doug.gus...@gs.com>> wrote: Hello, We are facing an issue with some of our applications that are submitting a high volume of jobs to Flink (we are using v1.9.2). We are observing that numerous jobs (in this case 44 out of 350+) fail with the same FlinkJobNotFoundException within a 45 second timeframe. From our client logs, this is the exception we can see: Calc Engine: Caused by: org.apache.flink.runtime.rest.util.RestClientException: [org.apache.flink.runtime.messages.FlinkJobNotFoundException: Could not find Flink job (d0991f0ae712a9df710aa03311a32c8c)] Calc Engine: at org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:389) Calc Engine: at org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:373) Calc Engine: at java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:952) Calc Engine: at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926) Calc Engine: at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442) Calc Engine: ... 3 more This is the first job to fail with the above exception. From the JobManager logs, we can see that the job goes to FINISHED State, and then we see the following exception: 2021-09-28 04:54:16,936 INFO [flink-akka.actor.default-dispatcher-28] org.apache.flink.runtime.executiongraph.ExecutionGraph- Job Flink Java Job at Tue Sep 28 04:48:21 EDT 2021 (d0991f0ae712a9df710aa03311a32c8c) switched from state RUNNING to FINISHED. 2021-09-28 04:54:16,937 INFO [flink-akka.actor.default-dispatcher-28] org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Job d0991f0ae712a9df710aa03311a32c8c reached globally terminal state FINISHED. 2021-09-28 04:54:16,939 INFO [flink-akka.actor.default-dispatcher-28] org.apache.flink.runtime.jobmaster.JobMaster - Stopping the JobMaster for job Flink Java Job at Tue Sep 28 04:48:21 EDT 2021(d0991f0ae712a9df710aa03311a32c8c). 2021-09-28 04:54:16,940 INFO [flink-akka.actor.default-dispatcher-39] org.apache.flink.yarn.YarnResourceManager - Disconnect job manager 0...@akka.tcp://fl...@d43723-714.dc.gs.com:44887/user/jobmanager_392<mailto:0...@akka.tcp://fl...@d43723-714.dc.gs.com:44887/user/jobmanager_392> for job d0991f0ae712a9df710aa03311a32c8c from the resource manager. 2021-09-28 04:54:18,256 ERROR [flink-akka.actor.default-dispatcher-91] org.apache.flink.runtime.rest.handler.job.JobExecutionResultHandler - Exception occurred in REST handler: org.apache.flink.runtime.messages.FlinkJobNo
Re: FlinkJobNotFoundException
I didn't receive any email. But we rather not do individual support. Please share the logs on the mailing list. This way, anyone is able to participate in the discussion. Best, Matthias On Wed, Sep 29, 2021 at 8:12 PM Gusick, Doug S wrote: > Hi Matthias, > > > > Thank you for getting back. We have been looking into upgrading to a newer > version, but have not completed full testing just yet. > > > > I was unable to find a previous error in the JM logs. You should have > received an email with details to a “lockbox”. I have uploaded the job > manager logs there. Please let me know if you need any more information. > > > > Thank you, > > Doug > > > > *From:* Matthias Pohl > *Sent:* Wednesday, September 29, 2021 12:00 PM > *To:* Gusick, Doug S [Engineering] > *Cc:* user@flink.apache.org; Erai, Rahul [Engineering] < > rahul.e...@ny.email.gs.com> > *Subject:* Re: FlinkJobNotFoundException > > > > Hi Doug, > > thanks for reaching out to the community. First of all, 1.9.2 is quite an > old Flink version. You might want to consider upgrading to a newer version. > The community only offers support for the two most-recent Flink versions. > Newer version might include fixes for your issue. > > > > But back to your actual problem: The logs you're providing only show that > some job switched into FINISHED state. Is there some error showing up > earlier in the logs which you might have missed? It would be helpful if you > could share the complete JobManager logs to get a better understanding of > what's going on. > > > > Best, > Matthias > > > > On Wed, Sep 29, 2021 at 3:47 PM Gusick, Doug S wrote: > > Hello, > > > > We are facing an issue with some of our applications that are submitting a > high volume of jobs to Flink (we are using v1.9.2). We are observing that > numerous jobs (in this case 44 out of 350+) fail with the same > FlinkJobNotFoundException within a 45 second timeframe. > > > > From our client logs, this is the exception we can see: > > > > Calc Engine: Caused by: > org.apache.flink.runtime.rest.util.RestClientException: > [org.apache.flink.runtime.messages.FlinkJobNotFoundException: Could not find > Flink job (d0991f0ae712a9df710aa03311a32c8c)] > > Calc Engine: at > org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:389) > > Calc Engine: at > org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:373) > > Calc Engine: at > java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:952) > > Calc Engine: at > java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926) > > Calc Engine: at > java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442) > > Calc Engine: ... 3 more > > > > > > This is the first job to fail with the above exception. From the > JobManager logs, we can see that the job goes to FINISHED State, and then > we see the following exception: > > > > 2021-09-28 04:54:16,936 INFO [flink-akka.actor.default-dispatcher-28] > org.apache.flink.runtime.executiongraph.ExecutionGraph- Job Flink > Java Job at Tue Sep 28 04:48:21 EDT 2021 (d0991f0ae712a9df710aa03311a32c8c) > switched from state RUNNING to FINISHED. > > 2021-09-28 04:54:16,937 INFO [flink-akka.actor.default-dispatcher-28] > org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Job > d0991f0ae712a9df710aa03311a32c8c reached globally terminal state FINISHED. > > 2021-09-28 04:54:16,939 INFO [flink-akka.actor.default-dispatcher-28] > org.apache.flink.runtime.jobmaster.JobMaster - Stopping > the JobMaster for job Flink Java Job at Tue Sep 28 04:48:21 EDT > 2021(d0991f0ae712a9df710aa03311a32c8c). > > 2021-09-28 04:54:16,940 INFO [flink-akka.actor.default-dispatcher-39] > org.apache.flink.yarn.YarnResourceManager - Disconnect > job manager > 0...@akka.tcp://fl...@d43723-714.dc.gs.com:44887/user/jobmanager_392 > for job d0991f0ae712a9df710aa03311a32c8c from the resource manager. > > 2021-09-28 04:54:18,256 ERROR [flink-akka.actor.default-dispatcher-91] > org.apache.flink.runtime.rest.handler.job.JobExecutionResultHandler - > Exception occurred in REST handler: > org.apache.flink.runtime.messages.FlinkJobNotFoundException: Could not find > Flink job (d0991f0ae712a9df710aa03311a32c8c) > > > > Here are the relevant logs from the TaskManager. We can see that the > JobLeaderService tries to reconnect to the job. Any ideas as to why it is > trying to reconnect?: > > > 2021-09-28 04:54:13,382 INFO [flink-akka.actor.def
RE: FlinkJobNotFoundException
Hi Matthias, Thank you for getting back. We have been looking into upgrading to a newer version, but have not completed full testing just yet. I was unable to find a previous error in the JM logs. You should have received an email with details to a “lockbox”. I have uploaded the job manager logs there. Please let me know if you need any more information. Thank you, Doug From: Matthias Pohl Sent: Wednesday, September 29, 2021 12:00 PM To: Gusick, Doug S [Engineering] Cc: user@flink.apache.org; Erai, Rahul [Engineering] Subject: Re: FlinkJobNotFoundException Hi Doug, thanks for reaching out to the community. First of all, 1.9.2 is quite an old Flink version. You might want to consider upgrading to a newer version. The community only offers support for the two most-recent Flink versions. Newer version might include fixes for your issue. But back to your actual problem: The logs you're providing only show that some job switched into FINISHED state. Is there some error showing up earlier in the logs which you might have missed? It would be helpful if you could share the complete JobManager logs to get a better understanding of what's going on. Best, Matthias On Wed, Sep 29, 2021 at 3:47 PM Gusick, Doug S mailto:doug.gus...@gs.com>> wrote: Hello, We are facing an issue with some of our applications that are submitting a high volume of jobs to Flink (we are using v1.9.2). We are observing that numerous jobs (in this case 44 out of 350+) fail with the same FlinkJobNotFoundException within a 45 second timeframe. From our client logs, this is the exception we can see: Calc Engine: Caused by: org.apache.flink.runtime.rest.util.RestClientException: [org.apache.flink.runtime.messages.FlinkJobNotFoundException: Could not find Flink job (d0991f0ae712a9df710aa03311a32c8c)] Calc Engine: at org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:389) Calc Engine: at org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:373) Calc Engine: at java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:952) Calc Engine: at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926) Calc Engine: at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442) Calc Engine: ... 3 more This is the first job to fail with the above exception. From the JobManager logs, we can see that the job goes to FINISHED State, and then we see the following exception: 2021-09-28 04:54:16,936 INFO [flink-akka.actor.default-dispatcher-28] org.apache.flink.runtime.executiongraph.ExecutionGraph- Job Flink Java Job at Tue Sep 28 04:48:21 EDT 2021 (d0991f0ae712a9df710aa03311a32c8c) switched from state RUNNING to FINISHED. 2021-09-28 04:54:16,937 INFO [flink-akka.actor.default-dispatcher-28] org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Job d0991f0ae712a9df710aa03311a32c8c reached globally terminal state FINISHED. 2021-09-28 04:54:16,939 INFO [flink-akka.actor.default-dispatcher-28] org.apache.flink.runtime.jobmaster.JobMaster - Stopping the JobMaster for job Flink Java Job at Tue Sep 28 04:48:21 EDT 2021(d0991f0ae712a9df710aa03311a32c8c). 2021-09-28 04:54:16,940 INFO [flink-akka.actor.default-dispatcher-39] org.apache.flink.yarn.YarnResourceManager - Disconnect job manager 0...@akka.tcp://fl...@d43723-714.dc.gs.com:44887/user/jobmanager_392<mailto:0...@akka.tcp://fl...@d43723-714.dc.gs.com:44887/user/jobmanager_392> for job d0991f0ae712a9df710aa03311a32c8c from the resource manager. 2021-09-28 04:54:18,256 ERROR [flink-akka.actor.default-dispatcher-91] org.apache.flink.runtime.rest.handler.job.JobExecutionResultHandler - Exception occurred in REST handler: org.apache.flink.runtime.messages.FlinkJobNotFoundException: Could not find Flink job (d0991f0ae712a9df710aa03311a32c8c) Here are the relevant logs from the TaskManager. We can see that the JobLeaderService tries to reconnect to the job. Any ideas as to why it is trying to reconnect?: 2021-09-28 04:54:13,382 INFO [flink-akka.actor.default-dispatcher-16] org.apache.flink.runtime.taskexecutor.TaskExecutor- Receive slot request b26c04706fd5aad03dfdca8691f1bf1c for job d0991f0ae712a9df710aa03311a32c8c from resource manager with leader id . 2021-09-28 04:54:13,383 INFO [flink-akka.actor.default-dispatcher-16] org.apache.flink.runtime.taskexecutor.JobLeaderService- Add job d0991f0ae712a9df710aa03311a32c8c for job leader monitoring. 2021-09-28 04:54:13,397 INFO [flink-akka.actor.default-dispatcher-16] org.apache.flink.runtime.taskexecutor.JobLeaderService- Successful registration at job manager akka.tcp://fl...@d43723-714.dc.gs.com:44887/user/jobmanager_392 for job d0991f0ae712a9df710aa03311a32c8c. 2021-09-28 04:54
Re: FlinkJobNotFoundException
Hi Doug, thanks for reaching out to the community. First of all, 1.9.2 is quite an old Flink version. You might want to consider upgrading to a newer version. The community only offers support for the two most-recent Flink versions. Newer version might include fixes for your issue. But back to your actual problem: The logs you're providing only show that some job switched into FINISHED state. Is there some error showing up earlier in the logs which you might have missed? It would be helpful if you could share the complete JobManager logs to get a better understanding of what's going on. Best, Matthias On Wed, Sep 29, 2021 at 3:47 PM Gusick, Doug S wrote: > Hello, > > > > We are facing an issue with some of our applications that are submitting a > high volume of jobs to Flink (we are using v1.9.2). We are observing that > numerous jobs (in this case 44 out of 350+) fail with the same > FlinkJobNotFoundException within a 45 second timeframe. > > > > From our client logs, this is the exception we can see: > > > > Calc Engine: Caused by: > org.apache.flink.runtime.rest.util.RestClientException: > [org.apache.flink.runtime.messages.FlinkJobNotFoundException: Could not find > Flink job (d0991f0ae712a9df710aa03311a32c8c)] > > Calc Engine: at > org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:389) > > Calc Engine: at > org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:373) > > Calc Engine: at > java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:952) > > Calc Engine: at > java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926) > > Calc Engine: at > java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442) > > Calc Engine: ... 3 more > > > > > > This is the first job to fail with the above exception. From the > JobManager logs, we can see that the job goes to FINISHED State, and then > we see the following exception: > > > > 2021-09-28 04:54:16,936 INFO [flink-akka.actor.default-dispatcher-28] > org.apache.flink.runtime.executiongraph.ExecutionGraph- Job Flink > Java Job at Tue Sep 28 04:48:21 EDT 2021 (d0991f0ae712a9df710aa03311a32c8c) > switched from state RUNNING to FINISHED. > > 2021-09-28 04:54:16,937 INFO [flink-akka.actor.default-dispatcher-28] > org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Job > d0991f0ae712a9df710aa03311a32c8c reached globally terminal state FINISHED. > > 2021-09-28 04:54:16,939 INFO [flink-akka.actor.default-dispatcher-28] > org.apache.flink.runtime.jobmaster.JobMaster - Stopping > the JobMaster for job Flink Java Job at Tue Sep 28 04:48:21 EDT > 2021(d0991f0ae712a9df710aa03311a32c8c). > > 2021-09-28 04:54:16,940 INFO [flink-akka.actor.default-dispatcher-39] > org.apache.flink.yarn.YarnResourceManager - Disconnect > job manager > 0...@akka.tcp://fl...@d43723-714.dc.gs.com:44887/user/jobmanager_392 > for job d0991f0ae712a9df710aa03311a32c8c from the resource manager. > > 2021-09-28 04:54:18,256 ERROR [flink-akka.actor.default-dispatcher-91] > org.apache.flink.runtime.rest.handler.job.JobExecutionResultHandler - > Exception occurred in REST handler: > org.apache.flink.runtime.messages.FlinkJobNotFoundException: Could not find > Flink job (d0991f0ae712a9df710aa03311a32c8c) > > > > Here are the relevant logs from the TaskManager. We can see that the > JobLeaderService tries to reconnect to the job. Any ideas as to why it is > trying to reconnect?: > > > 2021-09-28 04:54:13,382 INFO [flink-akka.actor.default-dispatcher-16] > org.apache.flink.runtime.taskexecutor.TaskExecutor- Receive slot > request b26c04706fd5aad03dfdca8691f1bf1c for job > d0991f0ae712a9df710aa03311a32c8c from resource manager with leader id > . > > 2021-09-28 04:54:13,383 INFO [flink-akka.actor.default-dispatcher-16] > org.apache.flink.runtime.taskexecutor.JobLeaderService- Add job > d0991f0ae712a9df710aa03311a32c8c for job leader monitoring. > > 2021-09-28 04:54:13,397 INFO [flink-akka.actor.default-dispatcher-16] > org.apache.flink.runtime.taskexecutor.JobLeaderService- Successful > registration at job manager > akka.tcp://fl...@d43723-714.dc.gs.com:44887/user/jobmanager_392 for job > d0991f0ae712a9df710aa03311a32c8c. > > 2021-09-28 04:54:13,397 INFO [flink-akka.actor.default-dispatcher-16] > org.apache.flink.runtime.taskexecutor.TaskExecutor- Establish > JobManager connection for job d0991f0ae712a9df710aa03311a32c8c. > > 2021-09-28 04:54:13,397 INFO [flink-akka.actor.default-dispatcher-
FlinkJobNotFoundException
Hello,
We are facing an issue with some of our applications that are submitting a high
volume of jobs to Flink (we are using v1.9.2). We are observing that numerous
jobs (in this case 44 out of 350+) fail with the same FlinkJobNotFoundException
within a 45 second timeframe.
>From our client logs, this is the exception we can see:
Calc Engine: Caused by: org.apache.flink.runtime.rest.util.RestClientException:
[org.apache.flink.runtime.messages.FlinkJobNotFoundException: Could not find
Flink job (d0991f0ae712a9df710aa03311a32c8c)]
Calc Engine: at
org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:389)
Calc Engine: at
org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:373)
Calc Engine: at
java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:952)
Calc Engine: at
java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926)
Calc Engine: at
java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
Calc Engine: ... 3 more
This is the first job to fail with the above exception. From the JobManager
logs, we can see that the job goes to FINISHED State, and then we see the
following exception:
2021-09-28 04:54:16,936 INFO [flink-akka.actor.default-dispatcher-28]
org.apache.flink.runtime.executiongraph.ExecutionGraph- Job Flink Java
Job at Tue Sep 28 04:48:21 EDT 2021 (d0991f0ae712a9df710aa03311a32c8c) switched
from state RUNNING to FINISHED.
2021-09-28 04:54:16,937 INFO [flink-akka.actor.default-dispatcher-28]
org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Job
d0991f0ae712a9df710aa03311a32c8c reached globally terminal state FINISHED.
2021-09-28 04:54:16,939 INFO [flink-akka.actor.default-dispatcher-28]
org.apache.flink.runtime.jobmaster.JobMaster - Stopping the
JobMaster for job Flink Java Job at Tue Sep 28 04:48:21 EDT
2021(d0991f0ae712a9df710aa03311a32c8c).
2021-09-28 04:54:16,940 INFO [flink-akka.actor.default-dispatcher-39]
org.apache.flink.yarn.YarnResourceManager - Disconnect job
manager
0...@akka.tcp://fl...@d43723-714.dc.gs.com:44887/user/jobmanager_392<mailto:0...@akka.tcp://fl...@d43723-714.dc.gs.com:44887/user/jobmanager_392>
for job d0991f0ae712a9df710aa03311a32c8c from the resource manager.
2021-09-28 04:54:18,256 ERROR [flink-akka.actor.default-dispatcher-91]
org.apache.flink.runtime.rest.handler.job.JobExecutionResultHandler -
Exception occurred in REST handler:
org.apache.flink.runtime.messages.FlinkJobNotFoundException: Could not find
Flink job (d0991f0ae712a9df710aa03311a32c8c)
Here are the relevant logs from the TaskManager. We can see that the
JobLeaderService tries to reconnect to the job. Any ideas as to why it is
trying to reconnect?:
2021-09-28 04:54:13,382 INFO [flink-akka.actor.default-dispatcher-16]
org.apache.flink.runtime.taskexecutor.TaskExecutor- Receive slot
request b26c04706fd5aad03dfdca8691f1bf1c for job
d0991f0ae712a9df710aa03311a32c8c from resource manager with leader id
.
2021-09-28 04:54:13,383 INFO [flink-akka.actor.default-dispatcher-16]
org.apache.flink.runtime.taskexecutor.JobLeaderService- Add job
d0991f0ae712a9df710aa03311a32c8c for job leader monitoring.
2021-09-28 04:54:13,397 INFO [flink-akka.actor.default-dispatcher-16]
org.apache.flink.runtime.taskexecutor.JobLeaderService- Successful
registration at job manager
akka.tcp://fl...@d43723-714.dc.gs.com:44887/user/jobmanager_392 for job
d0991f0ae712a9df710aa03311a32c8c.
2021-09-28 04:54:13,397 INFO [flink-akka.actor.default-dispatcher-16]
org.apache.flink.runtime.taskexecutor.TaskExecutor- Establish
JobManager connection for job d0991f0ae712a9df710aa03311a32c8c.
2021-09-28 04:54:13,397 INFO [flink-akka.actor.default-dispatcher-16]
org.apache.flink.runtime.taskexecutor.TaskExecutor- Offer reserved
slots to the leader of job d0991f0ae712a9df710aa03311a32c8c.
2021-09-28 04:54:13,405 INFO [CHAIN DataSource (settl_delivery_type_code |
DistCp | Sourcing Files) -> FlatMap (settl_delivery_type_code | DistCp |
Copying Batch Data) (1/1)] org.apache.flink.runtime.blob.BlobClient
- Downloading
d0991f0ae712a9df710aa03311a32c8c/p-54dfdc41d9a995e5b75eb9eb29bcac91725fc425-be2bc5df9ef6ce331e8019cf32eb222b
from d43723-714.dc.gs.com/10.175.239.171:38726
2021-09-28 04:54:16,942 INFO [flink-akka.actor.default-dispatcher-3]
org.apache.flink.runtime.taskexecutor.slot.TaskSlotTable - Free slot
TaskSlot(index:0, state:ACTIVE, resource profile:
ResourceProfile{cpuCores=1.7976931348623157E308, heapMemoryInMB=2147483647,
directMemoryInMB=2147483647, nativeMemoryInMB=2147483647,
networkMemoryInMB=2147483647, managedMemoryInMB=9318}, allocationId:
b26c04706fd5aad03dfdca8691f
FlinkJobNotFoundException????
mysql??MySQLlinux??
py4j.protocol.Py4JJavaError: An error occurred while calling o6.execute.
: java.util.concurrent.ExecutionException:
org.apache.flink.runtime.rest.util.RestClientException:
[org.apache.flink.runtime.rest.handler.RestHandlerException:
org.apache.flink.runtime.messages.FlinkJobNotFoundException: Could not find
Flink job (4d6b1273229e0e16fa433c652b5cb74d)
??
from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic,
CheckpointingMode
from pyflink.table import StreamTableEnvironment, EnvironmentSettings, DataTypes
batch_source_ddl = """
CREATE TABLE mh_source_tab (
lid INT,
dir INT,
posid BIGINT,
km BIGINT,
poleId BIGINT,
ts BIGINT,
rt Decimal(6,2),
time1 VARCHAR
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://**',
'table-name' = 'nj_mh_test', --??
'driver' = 'com.mysql.cj.jdbc.Driver',
'username' = 'root',
'password' = 'root',
--'sink.buffer-flush.max-rows' = '100',
'sink.buffer-flush.interval' = '1s'
)
warn_alarm_mh_ddl = """
CREATE TABLE warn_alarm_mh_sink (
lid INT,
dir INT,
posid BIGINT,
km BIGINT,
poleId BIGINT,
extremum Decimal(6,2),
PRIMARY KEY (lid,dir,posid,poleId) NOT ENFORCED
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://**',
'table-name' = 'warn_mh_alarm_result', --??
'driver' = 'com.mysql.cj.jdbc.Driver',
'username' = 'root',
'password' = 'root',
--'sink.buffer-flush.max-rows' = '100',
'sink.buffer-flush.interval' = '1s'
)
"""
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)
env_settings =
EnvironmentSettings.new_instance().use_blink_planner().in_streaming_mode().build()
t_env = StreamTableEnvironment.create(env, environment_settings=env_settings)
t_env.execute_sql(batch_source_ddl )
t_env.execute_sql(warn_alarm_mh_ddl)
def threshold_alarm(delta_thres):
source = t_env.from_path("mh_source_tab") \
.where("rt < -1")\
.group_by("lid, dir, posid, km, poleId")\
.select("lid, dir, posid, km, poleId, max(rt) as
max_rt, min(rt) as min_rt, max(rt)-min(rt) as extremum")\
.where("extremum "+str(delta_thres))\
.select("lid, dir, posid, km, poleId, extremum")
source.execute_insert("warn_alarm_mh_sink") \
.get_job_client() \
.get_job_execution_result() \
.result()
if __name__ == '__main__':
threshold_alarm(delta_thres=0.5)
FlinkJobNotFoundException????
mysql??MySQLlinux??
py4j.protocol.Py4JJavaError: An error occurred while calling o6.execute.
: java.util.concurrent.ExecutionException:
org.apache.flink.runtime.rest.util.RestClientException:
[org.apache.flink.runtime.rest.handler.RestHandlerException:
org.apache.flink.runtime.messages.FlinkJobNotFoundException: Could not find
Flink job (4d6b1273229e0e16fa433c652b5cb74d)
??
from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic,
CheckpointingMode
from pyflink.table import StreamTableEnvironment, EnvironmentSettings, DataTypes
batch_source_ddl = """
CREATE TABLE mh_source_tab (
lid INT,
dir INT,
posid BIGINT,
km BIGINT,
poleId BIGINT,
ts BIGINT,
rt Decimal(6,2),
time1 VARCHAR
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://**',
'table-name' = 'nj_mh_test', --??
'driver' = 'com.mysql.cj.jdbc.Driver',
'username' = 'root',
'password' = 'root',
--'sink.buffer-flush.max-rows' = '100',
'sink.buffer-flush.interval' = '1s'
)
warn_alarm_mh_ddl = """
CREATE TABLE warn_alarm_mh_sink (
lid INT,
dir INT,
posid BIGINT,
km BIGINT,
poleId BIGINT,
extremum Decimal(6,2),
PRIMARY KEY (lid,dir,posid,poleId) NOT ENFORCED
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://**',
'table-name' = 'warn_mh_alarm_result', --??
'driver' = 'com.mysql.cj.jdbc.Driver',
'username' = 'root',
'password' = 'root',
--'sink.buffer-flush.max-rows' = '100',
'sink.buffer-flush.interval' = '1s'
)
"""
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)
env_settings =
EnvironmentSettings.new_instance().use_blink_planner().in_streaming_mode().build()
t_env = StreamTableEnvironment.create(env, environment_settings=env_settings)
t_env.execute_sql(batch_source_ddl )
t_env.execute_sql(warn_alarm_mh_ddl)
def threshold_alarm(delta_thres):
source = t_env.from_path("mh_source_tab") \
.where("rt < -1")\
.group_by("lid, dir, posid, km, poleId")\
.select("lid, dir, posid, km, poleId, max(rt) as
max_rt, min(rt) as min_rt, max(rt)-min(rt) as extremum")\
.where("extremum "+str(delta_thres))\
.select("lid, dir, posid, km, poleId, extremum")
source.execute_insert("warn_alarm_mh_sink") \
.get_job_client() \
.get_job_execution_result() \
.result()
if __name__ == '__main__':
threshold_alarm(delta_thres=0.5)
FlinkJobNotFoundException????
mysql??MySQLlinux??
py4j.protocol.Py4JJavaError: An error occurred while calling o6.execute.
: java.util.concurrent.ExecutionException:
org.apache.flink.runtime.rest.util.RestClientException:
[org.apache.flink.runtime.rest.handler.RestHandlerException:
org.apache.flink.runtime.messages.FlinkJobNotFoundException: Could not find
Flink job (4d6b1273229e0e16fa433c652b5cb74d)
??
from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic,
CheckpointingMode
from pyflink.table import StreamTableEnvironment, EnvironmentSettings, DataTypes
batch_source_ddl = """
CREATE TABLE mh_source_tab (
lid INT,
dir INT,
posid BIGINT,
km BIGINT,
poleId BIGINT,
ts BIGINT,
rt Decimal(6,2),
time1 VARCHAR
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://**',
'table-name' = 'nj_mh_test', --??
'driver' = 'com.mysql.cj.jdbc.Driver',
'username' = 'root',
'password' = 'root',
--'sink.buffer-flush.max-rows' = '100',
'sink.buffer-flush.interval' = '1s'
)
warn_alarm_mh_ddl = """
CREATE TABLE warn_alarm_mh_sink (
lid INT,
dir INT,
posid BIGINT,
km BIGINT,
poleId BIGINT,
extremum Decimal(6,2),
PRIMARY KEY (lid,dir,posid,poleId) NOT ENFORCED
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://**',
'table-name' = 'warn_mh_alarm_result', --??
'driver' = 'com.mysql.cj.jdbc.Driver',
'username' = 'root',
'password' = 'root',
--'sink.buffer-flush.max-rows' = '100',
'sink.buffer-flush.interval' = '1s'
)
"""
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)
env_settings =
EnvironmentSettings.new_instance().use_blink_planner().in_streaming_mode().build()
t_env = StreamTableEnvironment.create(env, environment_settings=env_settings)
t_env.execute_sql(batch_source_ddl)
t_env.execute_sql(warn_alarm_mh_ddl)
def threshold_alarm(delta_thres):
source = t_env.from_path("mh_source_tab") \
.where("rt < -1")\
.group_by("lid, dir, posid, km, poleId")\
.select("lid, dir, posid, km, poleId, max(rt) as
max_rt, min(rt) as min_rt, max(rt)-min(rt) as extremum")\
.where("extremum "+str(delta_thres))\
.select("lid, dir, posid, km, poleId, extremum")
source.execute_insert("warn_alarm_mh_sink") \
.get_job_client() \
.get_job_execution_result() \
.result()
if __name__ == '__main__':
threshold_alarm(delta_thres=0.5)
FlinkJobNotFoundException????
mysql??MySQLlinux??
py4j.protocol.Py4JJavaError: An error occurred while calling o6.execute.
: java.util.concurrent.ExecutionException:
org.apache.flink.runtime.rest.util.RestClientException:
[org.apache.flink.runtime.rest.handler.RestHandlerException:
org.apache.flink.runtime.messages.FlinkJobNotFoundException: Could not find
Flink job (4d6b1273229e0e16fa433c652b5cb74d)
??
from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic,
CheckpointingMode
from pyflink.table import StreamTableEnvironment, EnvironmentSettings, DataTypes
batch_source_ddl = """
CREATE TABLE mh_source_tab (
lid INT,
dir INT,
posid BIGINT,
km BIGINT,
poleId BIGINT,
ts BIGINT,
rt Decimal(6,2),
time1 VARCHAR
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://**',
'table-name' = 'nj_mh_test', --??
'driver' = 'com.mysql.cj.jdbc.Driver',
'username' = 'root',
'password' = 'root',
--'sink.buffer-flush.max-rows' = '100',
'sink.buffer-flush.interval' = '1s'
)
warn_alarm_mh_ddl = """
CREATE TABLE warn_alarm_mh_sink (
lid INT,
dir INT,
posid BIGINT,
km BIGINT,
poleId BIGINT,
extremum Decimal(6,2),
PRIMARY KEY (lid,dir,posid,poleId) NOT ENFORCED
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://**',
'table-name' = 'warn_mh_alarm_result', --??
'driver' = 'com.mysql.cj.jdbc.Driver',
'username' = 'root',
'password' = 'root',
--'sink.buffer-flush.max-rows' = '100',
'sink.buffer-flush.interval' = '1s'
)
"""
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)
env_settings =
EnvironmentSettings.new_instance().use_blink_planner().in_streaming_mode().build()
t_env = StreamTableEnvironment.create(env, environment_settings=env_settings)
t_env.execute_sql(mh_source_tab)
t_env.execute_sql(warn_alarm_mh_ddl)
def threshold_alarm(delta_thres):
source = t_env.from_path("mh_source_tab") \
.where("rt < -1")\
.group_by("lid, dir, posid, km, poleId")\
.select("lid, dir, posid, km, poleId, max(rt) as
max_rt, min(rt) as min_rt, max(rt)-min(rt) as extremum")\
.where("extremum "+str(delta_thres))\
.select("lid, dir, posid, km, poleId, extremum")
source.execute_insert("warn_alarm_mh_sink") \
.get_job_client() \
.get_job_execution_result() \
.result()
if __name__ == '__main__':
threshold_alarm(delta_thres=0.5)
