给kafka吐数据时每条记录之间休眠1秒试试.
文档位置 https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/event_timestamps_watermarks.html https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/stream/operators/joining.html AssignerWithPeriodicWatermarks assigns timestamps and generates watermarks periodically (possibly depending on the stream elements, or purely based on processing time). The interval (every n milliseconds) in which the watermark will be generated is defined via ExecutionConfig.setAutoWatermarkInterval(...). The assigner’s getCurrentWatermark() method will be called each time, and a new watermark will be emitted if the returned watermark is non-null and larger than the previous watermark. Here we show two simple examples of timestamp assigners that use periodic watermark generation. Note that Flink ships with a BoundedOutOfOrdernessTimestampExtractor similar to the BoundedOutOfOrdernessGenerator shown below, which you can read about here. 在 2020/4/5 上午10:21，“忝忝向仧”<153488...@qq.com> 写入: Hi, 感觉有点不对，我试着用stream1输入: 1 tom1 1553503185000 1 tom2 1553503186000 1 tom3 1553503187000 1 tom4 1553503188000 1 tom_late 1553503185000 stream2输入: 1 jerry2 1553503186000 1 jerry3 1553503187000 1 jerry4 1553503188000 但是结果打印的是: currentTimeStamp: 1553503185000,Key:1,EventTime:1553503185000,前一条数据的水位线:0 4> (1,tom1,1553503185000) currentTimeStamp: 1553503186000,Key:1,EventTime:1553503186000,前一条数据的水位线:1553503185000 4> (1,tom2,1553503186000) currentTimeStamp: 1553503187000,Key:1,EventTime:1553503187000,前一条数据的水位线:1553503186000 4> (1,tom3,1553503187000) currentTimeStamp: 1553503188000,Key:1,EventTime:1553503188000,前一条数据的水位线:1553503187000 4> (1,tom4,1553503188000) currentTimeStamp: 1553503188000,Key:1,EventTime:1553503185000,前一条数据的水位线:1553503188000 4> (1,tom_late,1553503185000) currentTimeStamp: 1553503186000,Key:1,EventTime:1553503186000,前一条数据的水位线:0 3> (1,jerry2,1553503186000) currentTimeStamp: 1553503187000,Key:1,EventTime:1553503187000,前一条数据的水位线:1553503186000 3> (1,jerry3,1553503187000) currentTimeStamp: 1553503188000,Key:1,EventTime:1553503188000,前一条数据的水位线:1553503187000 3> (1,jerry4,1553503188000) 2> tom1=jerry2 2> tom1=jerry3 2> tom2=jerry2 2> tom2=jerry3 2> tom3=jerry2 2> tom3=jerry3 2> tom_late=jerry2 2> tom_late=jerry3 我第二个流的窗口是[1553503186000,1553503188000),怎么判断出stream1中那条迟到的是没过期的? 谢谢. ------------------ 原始邮件 ------------------ 发件人: "libenchao"<libenc...@gmail.com>; 发送时间: 2020年4月5日(星期天) 上午10:04 收件人: "user-zh"<user-zh@flink.apache.org>; 主题: Re: 回复： 回复： Flink双流Join问题 嗯，可以这么理解。 忝忝向仧 <153488...@qq.com> 于2020年4月4日周六 下午11:20写道： > 额，明白了，意思是说两个流情况下 > 比如，stream1里面晚来的那条 > 1 tom_late 1553503185000的水印是1553503188000 > 但是stream2里面,这条1,jerry1,1553503185000的水印是1553503185000 > 所以取最小的，因此还是会被打印? > 是这么理解么? > > > > > > > ------------------&nbsp;原始邮件&nbsp;------------------ > 发件人:&nbsp;"libenchao"<libenc...@gmail.com&gt;; > 发送时间:&nbsp;2020年4月4日(星期六) 晚上11:04 > 收件人:&nbsp;"user-zh"<user-zh@flink.apache.org&gt;; > > 主题:&nbsp;Re: 回复： 回复： Flink双流Join问题 > > > > 两个stream输入的场景，operator的watermark是取两者的最小值。 > 所以虽然这条数据在第一个流里面看起来已经是肯定迟到了，但是有可能看第二个流的watermark它还没有过期。 > > > 忝忝向仧 <153488...@qq.com&gt; 于2020年4月4日周六 下午10:42写道： > > &gt; Hi: > &gt; &amp;nbsp; &amp;nbsp; 还有个疑问,我特意构造了个乱序的消息,还是3秒一个窗口 > &gt; stream1: > &gt; 1 tom1 1553503185000 > &gt; 1 tom2 1553503186000 > &gt; 1 tom3 1553503187000 > &gt; 1 tom4 1553503188000 > &gt; 1 tom_late 1553503185000 > &gt; > &gt; > &gt; > &gt; stream2: > &gt; 1 jerry1 1553503185000 > &gt; 1 jerry2 1553503186000 > &gt; 1 jerry3 1553503187000 > &gt; 1 jerry4 1553503188000 > &gt; > &gt; > &gt; 我代码还是之前的代码，定义水印都是maxDelayAllowed = 0L;也就是没有设置最大延时时间. > &gt; 那么，tom_late这条乱序，在第一个窗口[1553503185000,1553503188000)内不应该被输出吧? > &gt; 但是结果还是输出了,这个是为什么? > &gt; > &gt; > &gt; currentTimeStamp: > 1553503185000,Key:1,EventTime:1553503185000,前一条数据的水位线:0 > &gt; 4&amp;gt; (1,tom1,1553503185000) > &gt; currentTimeStamp: > &gt; 1553503186000,Key:1,EventTime:1553503186000,前一条数据的水位线:1553503185000 > &gt; 4&amp;gt; (1,tom2,1553503186000) > &gt; currentTimeStamp: > &gt; 1553503187000,Key:1,EventTime:1553503187000,前一条数据的水位线:1553503186000 > &gt; 4&amp;gt; (1,tom3,1553503187000) > &gt; currentTimeStamp: > &gt; 1553503188000,Key:1,EventTime:1553503188000,前一条数据的水位线:1553503187000 > &gt; 4&amp;gt; (1,tom4,1553503188000) > &gt; currentTimeStamp: > &gt; 1553503188000,Key:1,EventTime:1553503185000,前一条数据的水位线:1553503188000 > &gt; 4&amp;gt; (1,tom_late,1553503185000) > &gt; currentTimeStamp: > &gt; 1553503189000,Key:2,EventTime:1553503189000,前一条数据的水位线:1553503188000 > &gt; 4&amp;gt; (2,tom5,1553503189000) > &gt; currentTimeStamp: > &gt; 1553503191000,Key:2,EventTime:1553503191000,前一条数据的水位线:1553503189000 > &gt; 4&amp;gt; (2,tom6,1553503191000) > &gt; currentTimeStamp: > &gt; 1553503192000,Key:2,EventTime:1553503192000,前一条数据的水位线:1553503191000 > &gt; 4&amp;gt; (2,tom7,1553503192000) > &gt; currentTimeStamp: > &gt; 1553503193000,Key:2,EventTime:1553503193000,前一条数据的水位线:1553503192000 > &gt; 4&amp;gt; (2,tom8,1553503193000) > &gt; currentTimeStamp: > 1553503185000,Key:1,EventTime:1553503185000,前一条数据的水位线:0 > &gt; 3&amp;gt; (1,jerry1,1553503185000) > &gt; currentTimeStamp: > &gt; 1553503186000,Key:1,EventTime:1553503186000,前一条数据的水位线:1553503185000 > &gt; 3&amp;gt; (1,jerry2,1553503186000) > &gt; currentTimeStamp: > &gt; 1553503187000,Key:1,EventTime:1553503187000,前一条数据的水位线:1553503186000 > &gt; 3&amp;gt; (1,jerry3,1553503187000) > &gt; currentTimeStamp: > &gt; 1553503188000,Key:1,EventTime:1553503188000,前一条数据的水位线:1553503187000 > &gt; 3&amp;gt; (1,jerry4,1553503188000) > &gt; 2&amp;gt; tom1=jerry1 > &gt; 2&amp;gt; tom1=jerry2 > &gt; 2&amp;gt; tom1=jerry3 > &gt; 2&amp;gt; tom2=jerry1 > &gt; 2&amp;gt; tom2=jerry2 > &gt; 2&amp;gt; tom2=jerry3 > &gt; 2&amp;gt; tom3=jerry1 > &gt; 2&amp;gt; tom3=jerry2 > &gt; 2&amp;gt; tom3=jerry3 > &gt; 2&amp;gt; tom_late=jerry1 > &gt; 2&amp;gt; tom_late=jerry2 > &gt; 2&amp;gt; tom_late=jerry3 > &gt; > &gt; > &gt; > &gt; > &gt; > &gt; > &gt; > &gt; > &gt; > &gt; > &gt; > &gt; > &gt; ------------------&amp;nbsp;原始邮件&amp;nbsp;------------------ > &gt; 发件人:&amp;nbsp;"Djeng Lee"<lee.ro...@gmail.com&amp;gt;; > &gt; 发送时间:&amp;nbsp;2020年4月4日(星期六) 晚上6:35 > &gt; 收件人:&amp;nbsp;"user-zh@flink.apache.org"<user-zh@flink.apache.org > &amp;gt;; > &gt; > &gt; 主题:&amp;nbsp;Re: 回复： 回复： Flink双流Join问题 > &gt; > &gt; > &gt; > &gt; 刚刚我说的不严谨 > &gt; > &gt; Start = 1000000055000 - (1000000055000 - 0 + 3000) % 3000 = > 1000000053000 > &gt; End = 1000000053000 + 3000 > &gt; > &gt; //源码位置，所以窗口开端并不是你传入首条记录的作为开端。窗口划分是从0时间戳切过来的。 > &gt; > &gt; 文档说明: > &gt; > https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/event_timestamps_watermarks.html > &gt; > &gt; 相关代码 > &gt; public static long getWindowStartWithOffset(long timestamp, long > offset, > &gt; long windowSize) { > &gt; &amp;nbsp;&nbsp; return timestamp - (timestamp - offset + windowSize) > % windowSize; > &gt; } > &gt; > &gt; 在 2020/4/4 下午6:30，“忝忝向仧”<153488...@qq.com&amp;gt; 写入: > &gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 额，你的意思是滚动3秒的窗口开始和结束应该是 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 1000000055000 % 3 得出结果再拿到[start,end). > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 比如1000000055000 % 3 > &gt; 的结果是1000000053000,那么窗口是[1000000053000,1000000056000) > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 是这么理解吧 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &gt; ------------------&amp;amp;nbsp;原始邮件&amp;amp;nbsp;------------------ > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 发件人:&amp;amp;nbsp;"lee.roval"< > lee.ro...@gmail.com&amp;amp;gt;; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 发送时间:&amp;amp;nbsp;2020年4月4日(星期六) > 晚上6:25 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 收件人:&amp;amp;nbsp;" > user-zh@flink.apache.org"< > &gt; user-zh@flink.apache.org&amp;amp;gt;; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 主题:&amp;amp;nbsp;Re: 回复： Flink双流Join问题 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 55000的窗口分配，是对windowSize 求模然后拿到start 和 > end。 不是从你首条记录开始算。 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 在 2020/4/4 下午6:23，“忝忝向仧”< > 153488...@qq.com&amp;amp;gt; 写入: > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; 对，我只是回复把前面那串省略了，没写. > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; [1000000053000,1000000056000),为什么是1000000053000开始? > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; 我第一条输入的是 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; 1,tom1,1000000055000 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; > ------------------&amp;amp;amp;nbsp;原始邮件&amp;amp;amp;nbsp;------------------ > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; 发件人:&amp;amp;amp;nbsp;"libenchao"<libenc...@gmail.com > &amp;amp;amp;gt;; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; 发送时间:&amp;amp;amp;nbsp;2020年4月4日(星期六) 晚上6:20 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; 收件人:&amp;amp;amp;nbsp;"user-zh"<user-zh@flink.apache.org > &amp;amp;amp;gt;; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; 主题:&amp;amp;amp;nbsp;Re: > &gt; Flink双流Join问题 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; 你的watermark不是56000，而是1000000056000吧。所以应该是[1000000053000, > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; 1000000056000)是一个窗口吧。 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; 忝忝向仧 <153488...@qq.com > &amp;amp;amp;gt; > &gt; 于2020年4月4日周六 下午6:16写道： > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; 下发新的? > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 4是一个kafka的source，3是另外一个kafka的source. > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; 如果按照3秒的一个窗口 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; watermark触发窗口的条件是watermark_time&amp;amp;amp;amp;gt;=window_endtime > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 也就是说[55000,57000)应该是一个窗口的. > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 我是这么理解的，但是结果56000后就输出了 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > currentTimeStamp:&amp;amp;amp;amp;nbsp;1000000055000,Key:1,EventTime:1000000055000,前一条数据的水位线:0 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 4&amp;amp;amp;amp;gt; (1,tom1,1000000055000) > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > currentTimeStamp:&amp;amp;amp;amp;nbsp;1000000056000,Key:1,EventTime:1000000056000,前一条数据的水位线:1000000055000 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 4&amp;amp;amp;amp;gt; (1,tom2,1000000056000) > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > currentTimeStamp:&amp;amp;amp;amp;nbsp;1000000055000,Key:1,EventTime:1000000055000,前一条数据的水位线:0 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 3&amp;amp;amp;amp;gt; (1,jerry1,1000000055000) > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > currentTimeStamp:&amp;amp;amp;amp;nbsp;1000000056000,Key:1,EventTime:1000000056000,前一条数据的水位线:1000000055000 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 3&amp;amp;amp;amp;gt; (1,jerry2,1000000056000) > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 2&amp;amp;amp;amp;gt; tom1=jerry1 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > ------------------&amp;amp;amp;amp;nbsp;原始邮件&amp;amp;amp;amp;nbsp;------------------ > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 发件人:&amp;amp;amp;amp;nbsp;"lee.roval"<lee.ro...@gmail.com > &amp;amp;amp;amp;gt;; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 发送时间:&amp;amp;amp;amp;nbsp;2020年4月4日(星期六) 晚上6:10 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 收件人:&amp;amp;amp;amp;nbsp;"user-zh@flink.apache.org"< > user-zh@flink.apache.org > &gt; &amp;amp;amp;amp;gt;; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 主题:&amp;amp;amp;amp;nbsp;Re: Flink双流Join问题 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 56000后不是下发新的watermark了嘛 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; 在 2020/4/4 > &gt; 下午5:57，“忝忝向仧”<153488...@qq.com&amp;amp;amp;amp;gt; 写入: > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; 各位好:&amp;amp;amp;amp;amp;nbsp; &amp;amp;amp;amp;amp;nbsp; > &amp;amp;amp;amp;amp;nbsp; > &gt; &amp;amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; Flink双流Join遇到一个问题，能否解释下，谢谢. > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;amp;amp;amp;amp;nbsp; &amp;amp;amp;amp;amp;nbsp; > &amp;amp;amp;amp;amp;nbsp; > &gt; &amp;amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; ds1和ds2分别读取kafka两个流数据，使用event time和watermark特性，3s的一个翻滚窗口，定义如下: > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;amp;amp;amp;amp;nbsp; &amp;amp;amp;amp;amp;nbsp; > &amp;amp;amp;amp;amp;nbsp; > &gt; &amp;amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 最后，join输出的时候，为什么触发窗口的数据第二条就触发了? > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;amp;amp;amp;amp;nbsp; &amp;amp;amp;amp;amp;nbsp; > &amp;amp;amp;amp;amp;nbsp; > &gt; &amp;amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > 按照水印的触发条件应该是watermark_time&amp;amp;amp;amp;amp;gt;=window_endtime.那么，这里应该是1000000057000这条数据来了后才会触发，但是结果却是56000就触发了.为什么? > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > 定义的代码如下: > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; DataStream<String&amp;amp;amp;amp;amp;gt; stream1 = env > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; .addSource(new > FlinkKafkaConsumer09<String&amp;amp;amp;amp;amp;gt;("stream1", > &gt; new > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; SimpleStringSchema(), properties).setStartFromLatest()) > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; .assignTimestampsAndWatermarks( > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; new > &gt; AssignerWithPeriodicWatermarks<String&amp;amp;amp;amp;amp;gt;() { > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; long > &gt; currentTimeStamp = 0L; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; long > &gt; maxDelayAllowed = 0L; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; long > &gt; currentWaterMark; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; @Nullable > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; @Override > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; public > &gt; Watermark getCurrentWatermark() { > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; currentWaterMark = currentTimeStamp-maxDelayAllowed; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; return new > &gt; Watermark(currentWaterMark); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; } > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; @Override > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; public long > &gt; extractTimestamp(String s, long l) { > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; String[] > &gt; arr= s.split(" "); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; long > &gt; timeStamp = Long.parseLong(arr[2]); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; currentTimeStamp = Math.max(timeStamp, currentTimeStamp); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; System.out.println("currentTimeStamp: " +&amp;amp;amp;amp;nbsp; > &gt; currentTimeStamp +",Key:" + > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; arr[0] + > &gt; ",EventTime:" + timeStamp + ",前一条数据的水位线:" + currentWaterMark); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; return > &gt; timeStamp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; } > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; } > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; ); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; > DataStream<Tuple3<String,String,String&amp;amp;amp;amp;amp;gt;&amp;amp;amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; ds1 = > &gt; stream1.map(new MapFunction<String, > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > Tuple3<String,String,String&amp;amp;amp;amp;amp;gt;&amp;amp;amp;amp;amp;gt;() > { > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; @Override > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; public Tuple3<String, > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; String,String&amp;amp;amp;amp;amp;gt; map(String s1) throws Exception > { > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; String[] > &gt; arr = s1.split(" "); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; return > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; Tuple3.of(arr[0],arr[1],arr[2]); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; } > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; }); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > ds1.print(); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; DataStream<String&amp;amp;amp;amp;amp;gt; stream2 = env > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; .addSource(new > FlinkKafkaConsumer09<String&amp;amp;amp;amp;amp;gt;("stream2", > &gt; new > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; SimpleStringSchema(), properties).setStartFromLatest()) > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; .assignTimestampsAndWatermarks( > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; new > &gt; AssignerWithPeriodicWatermarks<String&amp;amp;amp;amp;amp;gt;() { > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; long > &gt; currentTimeStamp = 0L; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; long > &gt; maxDelayAllowed = 0L; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; long > &gt; currentWaterMark; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; @Nullable > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; @Override > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; public > &gt; Watermark getCurrentWatermark() { > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; currentWaterMark = currentTimeStamp-maxDelayAllowed; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; return new > &gt; Watermark(currentWaterMark); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; } > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; @Override > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; public long > &gt; extractTimestamp(String s, long l) { > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; String[] > &gt; arr= s.split(" "); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; long > &gt; timeStamp = Long.parseLong(arr[2]); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; currentTimeStamp = Math.max(timeStamp, currentTimeStamp); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; System.out.println("currentTimeStamp: " +&amp;amp;amp;amp;nbsp; > &gt; currentTimeStamp +",Key:" + > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; arr[0] + > &gt; ",EventTime:" + timeStamp + ",前一条数据的水位线:" + currentWaterMark); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; return > &gt; timeStamp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; } > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; } > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; ); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; > DataStream<Tuple3<String,String,String&amp;amp;amp;amp;amp;gt;&amp;amp;amp;amp;amp;gt; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; ds2 > &gt; =stream2.map(new MapFunction<String, > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > Tuple3<String,String,String&amp;amp;amp;amp;amp;gt;&amp;amp;amp;amp;amp;gt;() > { > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; @Override > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; public > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; Tuple3<String,String,String&amp;amp;amp;amp;amp;gt; map(String s2) > throws > &gt; Exception { > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; String > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; [] arr = > &gt; s2.split(" "); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; return > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; Tuple3.of(arr[0],arr[1],arr[2]); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; } > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; }); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > ds2.print(); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; ds1.join(ds2).where(new KeySelector<Tuple3<String, > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > String,String&amp;amp;amp;amp;amp;gt;,String&amp;amp;amp;amp;amp;gt;() { > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; @Override > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; public String > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; getKey(Tuple3<String, String,String&amp;amp;amp;amp;amp;gt; value) > throws > &gt; Exception { > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; return > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; value.f0; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; } > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > }).equalTo(new > &gt; KeySelector<Tuple3<String, String > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; ,String&amp;amp;amp;amp;amp;gt;, String&amp;amp;amp;amp;amp;gt;() { > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; @Override > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; public String > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; getKey(Tuple3<String, String,String&amp;amp;amp;amp;amp;gt; value) > throws > &gt; Exception { > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; return > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; value.f0; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; } > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; }) > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; .window(TumblingEventTimeWindows.of(Time.seconds(3))) > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > .apply(new > &gt; JoinFunction<Tuple3<String, > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; String,String&amp;amp;amp;amp;amp;gt;, Tuple3<String, > &gt; String,String&amp;amp;amp;amp;amp;gt;, > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; String&amp;amp;amp;amp;amp;gt;() { > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; @Override > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; public String > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; join(Tuple3<String, String,String&amp;amp;amp;amp;amp;gt; value1, > Tuple3<String, > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; String,String&amp;amp;amp;amp;amp;gt; value2) throws Exception { > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; return > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; value1.f1 + > &gt; "=" + value2.f1; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; > &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; } > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > }).print(); > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > 结果如下: > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > currentTimeStamp: > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 1000000055000,Key:1,EventTime:1000000055000,前一条数据的水位线:0 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; 4&amp;amp;amp;amp;amp;gt; (1,tom1,1000000055000) > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > currentTimeStamp: > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 1000000056000,Key:1,EventTime:1000000056000,前一条数据的水位线:1000000055000 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; 4&amp;amp;amp;amp;amp;gt; (1,tom2,1000000056000) > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > currentTimeStamp: > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 1000000055000,Key:1,EventTime:1000000055000,前一条数据的水位线:0 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; 3&amp;amp;amp;amp;amp;gt; (1,jerry1,1000000055000) > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > currentTimeStamp: > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; 1000000056000,Key:1,EventTime:1000000056000,前一条数据的水位线:1000000055000 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; 3&amp;amp;amp;amp;amp;gt; (1,jerry2,1000000056000) > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; &amp;amp;amp;gt; > &gt; &amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp;&amp;amp;amp;amp;nbsp; > &gt; 2&amp;amp;amp;amp;amp;gt; tom1=jerry1 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; -- > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; Benchao Li > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; School of Electronics > &gt; Engineering and Computer Science, Peking University > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; Tel:+86-15650713730 > &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; Email: > &gt; libenc...@gmail.com; libenc...@pku.edu.cn > > > > -- > > Benchao Li > School of Electronics Engineering and Computer Science, Peking University > Tel:+86-15650713730 > Email: libenc...@gmail.com; libenc...@pku.edu.cn -- Benchao Li School of Electronics Engineering and Computer Science, Peking University Tel:+86-15650713730 Email: libenc...@gmail.com; libenc...@pku.edu.cn
