flink-user-zh mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "忝忝向仧" <153488...@qq.com>
Subject 回复: 回复: 回复: 回复: Flink双流Join问题
Date Sun, 05 Apr 2020 03:34:48 GMT
你的意思是说,在flink消费kafka时候,每条记录给个1秒休眠间隔?
比如,这里加个休眠1秒?
DataStream<String&gt; stream1 = env
        .addSource(new FlinkKafkaConsumer09<String&gt;("stream1", new SimpleStringSchema(), properties).
                setStartFromLatest());






------------------&nbsp;原始邮件&nbsp;------------------
发件人:&nbsp;"Djeng Lee"<lee.roval@gmail.com&gt;;
发送时间:&nbsp;2020年4月5日(星期天) 中午11:26
收件人:&nbsp;"user-zh@flink.apache.org"<user-zh@flink.apache.org&gt;;

主题:&nbsp;Re: 回复: 回复: 回复: Flink双流Join问题



给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,“忝忝向仧”<153488125@qq.com&gt; 写入:

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