Re: jobmanager 日志异常

2019-08-05 文章 Biao Liu 
你好，

> org.apache.flink.runtime.entrypoint.ClusterEntrypoint - RECEIVED
> SIGNAL 15: SIGTERM. Shutting down as requested.

这是收到了 signal 15 了 [1]，Wong 说得对，搜一下 yarn node manager 或者 yarn resource
manager 的 log

1. https://access.redhat.com/solutions/737033

Thanks,
Biao /'bɪ.aʊ/



On Tue, Aug 6, 2019 at 12:30 PM Wong Victor 
wrote:

> Hi，
>   可以查看一下jobmanager所在节点的yarn log，搜索一下对应的container为什么被kill；
>
> Regards
>
> On 2019/8/6, 11:40 AM, "戴嘉诚"  wrote:
>
> 大家好：
>
>
>
> 我的flink是部署在yarn上左session，今天早上jobmanager自动退出了，然后yarn把他重新拉起了，导致里面跑的job重新启动了，但是我查看日志，看到jobmanager的日志没有任何异常，同时jobmanager也没有长时间的full
> gc和频繁的gc，以下是jobmanager的日志：
> 就是在06:44分的是偶，日志上标记了收收到停止请求，然后jobmanager直接停止了...请问是由于什么原因导致的呢？
>
> 2019-08-06 06:43:58,891 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Completed
> > checkpoint 7843 for job e49624208fe771c4c9527799fd46f2a3 (5645215
> bytes in
> > 801 ms).
> > 2019-08-06 06:43:59,336 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Triggering
> > checkpoint 7852 @ 1565045039321 for job
> a9a7464ead55474bea6f42ed8e5de60f.
> > 2019-08-06 06:44:00,971 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Triggering
> > checkpoint 7852 @ 1565045040957 for job
> 79788b218e684cb31c1ca0fcc641e89f.
> > 2019-08-06 06:44:01,357 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Completed
> > checkpoint 7852 for job a9a7464ead55474bea6f42ed8e5de60f (25870658
> bytes in
> > 1806 ms).
> > 2019-08-06 06:44:02,887 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Completed
> > checkpoint 7852 for job 79788b218e684cb31c1ca0fcc641e89f (29798945
> bytes in
> > 1849 ms).
> > 2019-08-06 06:44:05,101 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Triggering
> > checkpoint 7852 @ 1565045045092 for job
> 03f3a0bd53c21f90f70ea01916dc9f78.
> > 2019-08-06 06:44:06,547 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Triggering
> > checkpoint 7844 @ 1565045046522 for job
> 486a1949d75863f823013d87b509d228.
> > 2019-08-06 06:44:07,311 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Completed
> > checkpoint 7844 for job 486a1949d75863f823013d87b509d228 (62458942
> bytes in
> > 736 ms).
> > 2019-08-06 06:44:07,506 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Completed
> > checkpoint 7852 for job 03f3a0bd53c21f90f70ea01916dc9f78 (105565032
> bytes
> > in 2366 ms).
> > 2019-08-06 06:44:08,087 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Triggering
> > checkpoint 7853 @ 1565045048055 for job
> 32783d371464265ef536454055ae6182.
> > 2019-08-06 06:44:09,626 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Checkpoint
> > 7050 of job 4b542195824ff7b7cdf749543fd368cb expired before
> completing.
> > 2019-08-06 06:44:09,647 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Triggering
> > checkpoint 7051 @ 1565045049626 for job
> 4b542195824ff7b7cdf749543fd368cb.
> > 2019-08-06 06:44:12,006 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Completed
> > checkpoint 7853 for job 32783d371464265ef536454055ae6182 (299599482
> bytes
> > in 3912 ms).
> > 2019-08-06 06:44:12,972 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Triggering
> > checkpoint 7853 @ 1565045052962 for job
> 16db5afe9a8cd7c6278030d5dec4c80c.
> > 2019-08-06 06:44:13,109 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Triggering
> > checkpoint 7853 @ 1565045053080 for job
> 9c1394a2d2ff47c7852eff9f1f932535.
> > 2019-08-06 06:44:16,779 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Completed
> > checkpoint 7853 for job 16db5afe9a8cd7c6278030d5dec4c80c (152643149
> bytes
> > in 3666 ms).
> > 2019-08-06 06:44:18,598 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Completed
> > checkpoint 7828 for job 8df2b47f2a4c1ba0f7019ee5989f6e71 (837558245
> bytes
> > in 23472 ms).
> > 2019-08-06 06:44:19,193 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Completed
> > checkpoint 7853 for job 9c1394a2d2ff47c7852eff9f1f932535 (594628825
> bytes
> > in 6067 ms).
> > 2019-08-06 06:44:19,238 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Completed
> > checkpoint 5855 for job 108ce7f6f5f3e76b12fad9dbdbc8feba (45917615
> bytes in
> > 61819 ms).
> > 2019-08-06 06:44:19,248 INFO
> >  org.apache.flink.runtime.checkpoint.CheckpointCoordinator -
> Triggering
> > checkpoint 5856 @ 1565045059238

Re: Re: Flink RocksDBStateBackend 问题

2019-08-05 文章 Yun Tang 
@lvwenyuan
首先需要明确的一点是，你这里的“FileSystem”指的究竟是checkpoint时存储数据的file 
system，还是FsStateBackend，建议下次提问前可以把需要咨询的内容表述清楚一些。

  *   如果指的是存储checkpoint数据的远程file system，在incremental 
checkpoint场景下，这些数据与RocksDB的创建checkpoint时刷写到本地的sst文件和meta文件是二进制相同的，只是文件名会重命名。如果是savepoint或者全量checkpoint场景下，这些数据是RocksDB中逐个有效entry的序列化内容。
  *   
如果指的是FsStateBackend，对于Flink而言存储的数据内容在逻辑上肯定都是一样的，否则就不符合语义了。但是二者在数据存储格式上是有区别的。FsStateBackend所创建的HeapKeyedStateBackend的数据内容都是存储在Java
 heap内的，基本数据格式是StateTable[1]和其中存储数据的StateMap[2]。而RocksDB存储的数据主要是RocksDB 
native内存中的writer buffer(memtable)，block 
cache，index[3]和已经刷写到磁盘上默认采用snappy压缩的不可变sst文件构成。

[1] 
https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/StateTable.java
[2] 
https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/StateMap.java
[3] https://github.com/facebook/rocksdb/wiki/Memory-usage-in-RocksDB


祝好
唐云


From: 戴嘉诚 
Sent: Tuesday, August 6, 2019 12:01
To: user-zh@flink.apache.org 
Subject: 答复: Re: Flink RocksDBStateBackend 问题


不是，文档上有说，filesystem是会把正在运行的数据存储在tm的内存中，然后触发checkpoint后，才会写入文件系统上，而rocksdb是直接把运行中的数据写到了rocksdb上，看样子是不占用运行中的tm的内存。

 
https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/state/state_backends.html#the-fsstatebackend

`The FsStateBackend holds in-flight data in the TaskManager’s memory. Upon 
checkpointing, it writes state snapshots into files in the configured file 
system and directory. Minimal metadata is stored in the JobManager’s memory 
(or, in high-availability mode, in the metadata checkpoint).`

发件人: athlon...@gmail.com
发送时间: 2019年8月6日 11:53
收件人: user-zh
主题: Re: Re: Flink RocksDBStateBackend 问题

你说的是memsystem的状态数据存在jm内存中的filesystem是存到文件系统上的



athlon...@gmail.com
发件人： 戴嘉诚
发送时间： 2019-08-06 11:42
收件人： user-zh
主题： Re: Flink RocksDBStateBackend 问题
FileSystem 我记得是存储的大小是不能超过tm的内存还是jm的内存，而rocksdb上存储的数据是可以无限的，不过相对来说，
FileSystem的吞吐就会比rocksdb会高
lvwenyuan  于2019年8月6日周二 上午11:39写道：
> 请教各位：
>RocksDBStateBackend
> 中，rocksdb上存储的内如和FileSystem上存储的数据内容是一样的？如果不一样，那么分别是什么呢？感谢回答
>
>
>
>

Re: jobmanager 日志异常

2019-08-05 文章 Wong Victor 
Hi，
  可以查看一下jobmanager所在节点的yarn log，搜索一下对应的container为什么被kill；

Regards

On 2019/8/6, 11:40 AM, "戴嘉诚"  wrote:

大家好：



我的flink是部署在yarn上左session，今天早上jobmanager自动退出了，然后yarn把他重新拉起了，导致里面跑的job重新启动了，但是我查看日志，看到jobmanager的日志没有任何异常，同时jobmanager也没有长时间的full
gc和频繁的gc，以下是jobmanager的日志：
就是在06:44分的是偶，日志上标记了收收到停止请求，然后jobmanager直接停止了...请问是由于什么原因导致的呢？

2019-08-06 06:43:58,891 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7843 for job e49624208fe771c4c9527799fd46f2a3 (5645215 bytes in
> 801 ms).
> 2019-08-06 06:43:59,336 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - 
Triggering
> checkpoint 7852 @ 1565045039321 for job a9a7464ead55474bea6f42ed8e5de60f.
> 2019-08-06 06:44:00,971 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - 
Triggering
> checkpoint 7852 @ 1565045040957 for job 79788b218e684cb31c1ca0fcc641e89f.
> 2019-08-06 06:44:01,357 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7852 for job a9a7464ead55474bea6f42ed8e5de60f (25870658 bytes 
in
> 1806 ms).
> 2019-08-06 06:44:02,887 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7852 for job 79788b218e684cb31c1ca0fcc641e89f (29798945 bytes 
in
> 1849 ms).
> 2019-08-06 06:44:05,101 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - 
Triggering
> checkpoint 7852 @ 1565045045092 for job 03f3a0bd53c21f90f70ea01916dc9f78.
> 2019-08-06 06:44:06,547 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - 
Triggering
> checkpoint 7844 @ 1565045046522 for job 486a1949d75863f823013d87b509d228.
> 2019-08-06 06:44:07,311 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7844 for job 486a1949d75863f823013d87b509d228 (62458942 bytes 
in
> 736 ms).
> 2019-08-06 06:44:07,506 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7852 for job 03f3a0bd53c21f90f70ea01916dc9f78 (105565032 bytes
> in 2366 ms).
> 2019-08-06 06:44:08,087 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - 
Triggering
> checkpoint 7853 @ 1565045048055 for job 32783d371464265ef536454055ae6182.
> 2019-08-06 06:44:09,626 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - 
Checkpoint
> 7050 of job 4b542195824ff7b7cdf749543fd368cb expired before completing.
> 2019-08-06 06:44:09,647 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - 
Triggering
> checkpoint 7051 @ 1565045049626 for job 4b542195824ff7b7cdf749543fd368cb.
> 2019-08-06 06:44:12,006 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7853 for job 32783d371464265ef536454055ae6182 (299599482 bytes
> in 3912 ms).
> 2019-08-06 06:44:12,972 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - 
Triggering
> checkpoint 7853 @ 1565045052962 for job 16db5afe9a8cd7c6278030d5dec4c80c.
> 2019-08-06 06:44:13,109 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - 
Triggering
> checkpoint 7853 @ 1565045053080 for job 9c1394a2d2ff47c7852eff9f1f932535.
> 2019-08-06 06:44:16,779 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7853 for job 16db5afe9a8cd7c6278030d5dec4c80c (152643149 bytes
> in 3666 ms).
> 2019-08-06 06:44:18,598 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7828 for job 8df2b47f2a4c1ba0f7019ee5989f6e71 (837558245 bytes
> in 23472 ms).
> 2019-08-06 06:44:19,193 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7853 for job 9c1394a2d2ff47c7852eff9f1f932535 (594628825 bytes
> in 6067 ms).
> 2019-08-06 06:44:19,238 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 5855 for job 108ce7f6f5f3e76b12fad9dbdbc8feba (45917615 bytes 
in
> 61819 ms).
> 2019-08-06 06:44:19,248 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - 
Triggering
> checkpoint 5856 @ 1565045059238 for job 108ce7f6f5f3e76b12fad9dbdbc8feba.
> 2019-08-06 06:44:22,092 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - 
Triggering
> checkpoint 7802 @ 1565045062084 for job 430689e0f202fcb29ce9d6403e6825f9.
> 2019-08-06 06:44:22,838 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 2940 for job fea51fd74006de69e265adc13e802229 (122562953 bytes
> in 174336 ms).
> 2019-08-06 06:44:22,888 INFO
>

Re: Re: Flink RocksDBStateBackend 问题

2019-08-05 文章 athlon...@gmail.com 
你说的是memsystem的状态数据存在jm内存中的filesystem是存到文件系统上的



athlon...@gmail.com
 
发件人： 戴嘉诚
发送时间： 2019-08-06 11:42
收件人： user-zh
主题： Re: Flink RocksDBStateBackend 问题
FileSystem 我记得是存储的大小是不能超过tm的内存还是jm的内存，而rocksdb上存储的数据是可以无限的，不过相对来说，
FileSystem的吞吐就会比rocksdb会高
 
lvwenyuan  于2019年8月6日周二 上午11:39写道：
 
> 请教各位：
>RocksDBStateBackend
> 中，rocksdb上存储的内如和FileSystem上存储的数据内容是一样的？如果不一样，那么分别是什么呢？感谢回答
>
>
>
>

Re: Flink RocksDBStateBackend 问题

2019-08-05 文章 戴嘉诚 
FileSystem 我记得是存储的大小是不能超过tm的内存还是jm的内存，而rocksdb上存储的数据是可以无限的，不过相对来说，
FileSystem的吞吐就会比rocksdb会高

lvwenyuan  于2019年8月6日周二 上午11:39写道：

> 请教各位：
>RocksDBStateBackend
> 中，rocksdb上存储的内如和FileSystem上存储的数据内容是一样的？如果不一样，那么分别是什么呢？感谢回答
>
>
>
>

jobmanager 日志异常

2019-08-05 文章 戴嘉诚 
大家好：


我的flink是部署在yarn上左session，今天早上jobmanager自动退出了，然后yarn把他重新拉起了，导致里面跑的job重新启动了，但是我查看日志，看到jobmanager的日志没有任何异常，同时jobmanager也没有长时间的full
gc和频繁的gc，以下是jobmanager的日志：
就是在06:44分的是偶，日志上标记了收收到停止请求，然后jobmanager直接停止了...请问是由于什么原因导致的呢？

2019-08-06 06:43:58,891 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7843 for job e49624208fe771c4c9527799fd46f2a3 (5645215 bytes in
> 801 ms).
> 2019-08-06 06:43:59,336 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering
> checkpoint 7852 @ 1565045039321 for job a9a7464ead55474bea6f42ed8e5de60f.
> 2019-08-06 06:44:00,971 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering
> checkpoint 7852 @ 1565045040957 for job 79788b218e684cb31c1ca0fcc641e89f.
> 2019-08-06 06:44:01,357 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7852 for job a9a7464ead55474bea6f42ed8e5de60f (25870658 bytes in
> 1806 ms).
> 2019-08-06 06:44:02,887 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7852 for job 79788b218e684cb31c1ca0fcc641e89f (29798945 bytes in
> 1849 ms).
> 2019-08-06 06:44:05,101 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering
> checkpoint 7852 @ 1565045045092 for job 03f3a0bd53c21f90f70ea01916dc9f78.
> 2019-08-06 06:44:06,547 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering
> checkpoint 7844 @ 1565045046522 for job 486a1949d75863f823013d87b509d228.
> 2019-08-06 06:44:07,311 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7844 for job 486a1949d75863f823013d87b509d228 (62458942 bytes in
> 736 ms).
> 2019-08-06 06:44:07,506 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7852 for job 03f3a0bd53c21f90f70ea01916dc9f78 (105565032 bytes
> in 2366 ms).
> 2019-08-06 06:44:08,087 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering
> checkpoint 7853 @ 1565045048055 for job 32783d371464265ef536454055ae6182.
> 2019-08-06 06:44:09,626 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Checkpoint
> 7050 of job 4b542195824ff7b7cdf749543fd368cb expired before completing.
> 2019-08-06 06:44:09,647 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering
> checkpoint 7051 @ 1565045049626 for job 4b542195824ff7b7cdf749543fd368cb.
> 2019-08-06 06:44:12,006 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7853 for job 32783d371464265ef536454055ae6182 (299599482 bytes
> in 3912 ms).
> 2019-08-06 06:44:12,972 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering
> checkpoint 7853 @ 1565045052962 for job 16db5afe9a8cd7c6278030d5dec4c80c.
> 2019-08-06 06:44:13,109 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering
> checkpoint 7853 @ 1565045053080 for job 9c1394a2d2ff47c7852eff9f1f932535.
> 2019-08-06 06:44:16,779 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7853 for job 16db5afe9a8cd7c6278030d5dec4c80c (152643149 bytes
> in 3666 ms).
> 2019-08-06 06:44:18,598 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7828 for job 8df2b47f2a4c1ba0f7019ee5989f6e71 (837558245 bytes
> in 23472 ms).
> 2019-08-06 06:44:19,193 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7853 for job 9c1394a2d2ff47c7852eff9f1f932535 (594628825 bytes
> in 6067 ms).
> 2019-08-06 06:44:19,238 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 5855 for job 108ce7f6f5f3e76b12fad9dbdbc8feba (45917615 bytes in
> 61819 ms).
> 2019-08-06 06:44:19,248 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering
> checkpoint 5856 @ 1565045059238 for job 108ce7f6f5f3e76b12fad9dbdbc8feba.
> 2019-08-06 06:44:22,092 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering
> checkpoint 7802 @ 1565045062084 for job 430689e0f202fcb29ce9d6403e6825f9.
> 2019-08-06 06:44:22,838 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 2940 for job fea51fd74006de69e265adc13e802229 (122562953 bytes
> in 174336 ms).
> 2019-08-06 06:44:22,888 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering
> checkpoint 2941 @ 1565045062838 for job fea51fd74006de69e265adc13e802229.
> 2019-08-06 06:44:24,348 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering
> checkpoint 613 @ 1565045064328 for job 5a75d77312f29c714af0a2994f0e8b1a.
> 2019-08-06 06:44:25,327 INFO
>  org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
> checkpoint 7802 for job 430689e0f202fcb29ce9d6403e6825f9 (358649788 bytes
> in

flink 动态表输出问题

2019-08-05 文章 金圣哲 
各位Flink社区大佬，
 你们好:

请教一下:

问题是 flink sql实现动态表之后,想基于动态表进行查询定时输出, 各位大神有什么实现的思路吗感激不尽.

SQL 1: tableEnv.sqlQuery("select date_id, id, latest(status, utime) as status, 
latest(user_id, utime) as user_id from waybillAppendTable group by date_id, 
id");

查询SQL2:tableEnv.sqlQuery("select date_id, user_id, status as status 
,count(status) as status_count from " + waybillTbable + "  group by 
date_id,status,user_id”);

希望能每分钟输出查询SQL2的结果呢!!!


感激不尽