flink-user-zh mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Djeng Lee <lee.ro...@gmail.com>
Subject Re: 回复: Flink双流Join问题
Date Sat, 04 Apr 2020 10:25:14 GMT
55000的窗口分配,是对windowSize 求模然后拿到start 和 end。 不是从你首条记录开始算。

在 2020/4/4 下午6:23,“忝忝向仧”<153488125@qq.com> 写入:

    对,我只是回复把前面那串省略了,没写.
    [1000000053000,1000000056000),为什么是1000000053000开始?
    我第一条输入的是
    1,tom1,1000000055000
    
    
    ------------------&nbsp;原始邮件&nbsp;------------------
    发件人:&nbsp;"libenchao"<libenchao@gmail.com&gt;;
    发送时间:&nbsp;2020年4月4日(星期六) 晚上6:20
    收件人:&nbsp;"user-zh"<user-zh@flink.apache.org&gt;;
    
    主题:&nbsp;Re: Flink双流Join问题
    
    
    
    你的watermark不是56000,而是1000000056000吧。所以应该是[1000000053000,
    1000000056000)是一个窗口吧。
    
    忝忝向仧 <153488125@qq.com&gt; 于2020年4月4日周六 下午6:16写道:
    
    &gt; 下发新的?
    &gt;
    &gt;
    &gt; 4是一个kafka的source,3是另外一个kafka的source.
    &gt; 如果按照3秒的一个窗口
    &gt; watermark触发窗口的条件是watermark_time&amp;gt;=window_endtime
    &gt;
    &gt;
    &gt; 也就是说[55000,57000)应该是一个窗口的.
    &gt;
    &gt;
    &gt; 我是这么理解的,但是结果56000后就输出了
    &gt;
    &gt;
    &gt;
    &gt; currentTimeStamp:&amp;nbsp;1000000055000,Key:1,EventTime:1000000055000,前一条数据的水位线:0
    &gt; 4&amp;gt; (1,tom1,1000000055000)
    &gt;
    &gt; currentTimeStamp:&amp;nbsp;1000000056000,Key:1,EventTime:1000000056000,前一条数据的水位线:1000000055000
    &gt; 4&amp;gt; (1,tom2,1000000056000)
    &gt;
    &gt; currentTimeStamp:&amp;nbsp;1000000055000,Key:1,EventTime:1000000055000,前一条数据的水位线:0
    &gt; 3&amp;gt; (1,jerry1,1000000055000)
    &gt;
    &gt; currentTimeStamp:&amp;nbsp;1000000056000,Key:1,EventTime:1000000056000,前一条数据的水位线:1000000055000
    &gt; 3&amp;gt; (1,jerry2,1000000056000)
    &gt; 2&amp;gt; tom1=jerry1
    &gt;
    &gt;
    &gt;
    &gt;
    &gt;
    &gt;
    &gt;
    &gt; ------------------&amp;nbsp;原始邮件&amp;nbsp;------------------
    &gt; 发件人:&amp;nbsp;"lee.roval"<lee.roval@gmail.com&amp;gt;;
    &gt; 发送时间:&amp;nbsp;2020年4月4日(星期六) 晚上6:10
    &gt; 收件人:&amp;nbsp;"user-zh@flink.apache.org"<user-zh@flink.apache.org&amp;gt;;
    &gt;
    &gt; 主题:&amp;nbsp;Re: Flink双流Join问题
    &gt;
    &gt;
    &gt;
    &gt; 56000后不是下发新的watermark了嘛
    &gt;
    &gt; 在 2020/4/4 下午5:57,“忝忝向仧”<153488125@qq.com&amp;gt;
写入:
    &gt;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 各位好:&amp;amp;nbsp; &amp;amp;nbsp;
&amp;amp;nbsp; &amp;amp;nbsp;
    &gt; Flink双流Join遇到一个问题,能否解释下,谢谢.
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp;
&amp;amp;nbsp; &amp;amp;nbsp;
    &gt; ds1和ds2分别读取kafka两个流数据,使用event time和watermark特性,3s的一个翻滚窗口,定义如下:
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp;
&amp;amp;nbsp; &amp;amp;nbsp;
    &gt; 最后,join输出的时候,为什么触发窗口的数据第二条就触发了?
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp;
&amp;amp;nbsp; &amp;amp;nbsp;
    &gt; 按照水印的触发条件应该是watermark_time&amp;amp;gt;=window_endtime.那么,这里应该是1000000057000这条数据来了后才会触发,但是结果却是56000就触发了.为什么?
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 定义的代码如下:
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; DataStream<String&amp;amp;gt;
stream1 = env
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; .addSource(new FlinkKafkaConsumer09<String&amp;amp;gt;("stream1", new
    &gt; SimpleStringSchema(), properties).setStartFromLatest())
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; .assignTimestampsAndWatermarks(
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; new AssignerWithPeriodicWatermarks<String&amp;amp;gt;() {
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; long currentTimeStamp = 0L;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; long maxDelayAllowed = 0L;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; long currentWaterMark;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; @Nullable
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; @Override
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; public Watermark getCurrentWatermark() {
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; currentWaterMark = currentTimeStamp-maxDelayAllowed;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; return new Watermark(currentWaterMark);
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; }
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; @Override
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; public long extractTimestamp(String s, long l) {
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; String[] arr= s.split(" ");
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; long timeStamp = Long.parseLong(arr[2]);
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; currentTimeStamp = Math.max(timeStamp, currentTimeStamp);
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; System.out.println("currentTimeStamp: " +&amp;nbsp; currentTimeStamp +",Key:"
+
    &gt; arr[0] + ",EventTime:" + timeStamp + ",前一条数据的水位线:" + currentWaterMark);
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; return timeStamp;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; }
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; }
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
);
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; DataStream<Tuple3<String,String,String&amp;amp;gt;&amp;amp;gt;
    &gt; ds1 = stream1.map(new MapFunction<String,
    &gt; Tuple3<String,String,String&amp;amp;gt;&amp;amp;gt;() {
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
@Override
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
public Tuple3<String,
    &gt; String,String&amp;amp;gt; map(String s1) throws Exception {
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; String[] arr = s1.split(" ");
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
return
    &gt; Tuple3.of(arr[0],arr[1],arr[2]);
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
}
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; });
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; ds1.print();
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; DataStream<String&amp;amp;gt;
stream2 = env
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; .addSource(new FlinkKafkaConsumer09<String&amp;amp;gt;("stream2", new
    &gt; SimpleStringSchema(), properties).setStartFromLatest())
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; .assignTimestampsAndWatermarks(
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; new AssignerWithPeriodicWatermarks<String&amp;amp;gt;() {
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; long currentTimeStamp = 0L;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; long maxDelayAllowed = 0L;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; long currentWaterMark;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; @Nullable
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; @Override
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; public Watermark getCurrentWatermark() {
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; currentWaterMark = currentTimeStamp-maxDelayAllowed;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; return new Watermark(currentWaterMark);
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; }
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; @Override
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; public long extractTimestamp(String s, long l) {
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; String[] arr= s.split(" ");
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; long timeStamp = Long.parseLong(arr[2]);
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; currentTimeStamp = Math.max(timeStamp, currentTimeStamp);
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; System.out.println("currentTimeStamp: " +&amp;nbsp; currentTimeStamp +",Key:"
+
    &gt; arr[0] + ",EventTime:" + timeStamp + ",前一条数据的水位线:" + currentWaterMark);
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; return timeStamp;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; }
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; }
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
);
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; DataStream<Tuple3<String,String,String&amp;amp;gt;&amp;amp;gt;
    &gt; ds2 =stream2.map(new MapFunction<String,
    &gt; Tuple3<String,String,String&amp;amp;gt;&amp;amp;gt;() {
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
@Override
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
public
    &gt; Tuple3<String,String,String&amp;amp;gt; map(String s2) throws Exception
{
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
String
    &gt; [] arr = s2.split(" ");
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
return
    &gt; Tuple3.of(arr[0],arr[1],arr[2]);
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
}
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; });
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; ds2.print();
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; ds1.join(ds2).where(new KeySelector<Tuple3<String,
    &gt; String,String&amp;amp;gt;,String&amp;amp;gt;() {
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
@Override
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
public String
    &gt; getKey(Tuple3<String, String,String&amp;amp;gt; value) throws Exception
{
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
return
    &gt; value.f0;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
}
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; }).equalTo(new KeySelector<Tuple3<String,
String
    &gt; ,String&amp;amp;gt;, String&amp;amp;gt;() {
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
@Override
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
public String
    &gt; getKey(Tuple3<String, String,String&amp;amp;gt; value) throws Exception
{
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
return
    &gt; value.f0;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
}
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; })
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; .window(TumblingEventTimeWindows.of(Time.seconds(3)))
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; .apply(new JoinFunction<Tuple3<String,
    &gt; String,String&amp;amp;gt;, Tuple3<String, String,String&amp;amp;gt;,
    &gt; String&amp;amp;gt;() {
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
@Override
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
public String
    &gt; join(Tuple3<String, String,String&amp;amp;gt; value1, Tuple3<String,
    &gt; String,String&amp;amp;gt; value2) throws Exception {
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
return
    &gt; value1.f1 + "=" + value2.f1;
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
}
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; }).print();
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 结果如下:
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; currentTimeStamp:
    &gt; 1000000055000,Key:1,EventTime:1000000055000,前一条数据的水位线:0
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 4&amp;amp;gt; (1,tom1,1000000055000)
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; currentTimeStamp:
    &gt; 1000000056000,Key:1,EventTime:1000000056000,前一条数据的水位线:1000000055000
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 4&amp;amp;gt; (1,tom2,1000000056000)
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; currentTimeStamp:
    &gt; 1000000055000,Key:1,EventTime:1000000055000,前一条数据的水位线:0
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 3&amp;amp;gt; (1,jerry1,1000000055000)
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; currentTimeStamp:
    &gt; 1000000056000,Key:1,EventTime:1000000056000,前一条数据的水位线:1000000055000
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 3&amp;amp;gt; (1,jerry2,1000000056000)
    &gt; &amp;nbsp;&amp;nbsp;&amp;nbsp; 2&amp;amp;gt; tom1=jerry1
    
    
    
    -- 
    
    Benchao Li
    School of Electronics Engineering and Computer Science, Peking University
    Tel:+86-15650713730
    Email: libenchao@gmail.com; libenchao@pku.edu.cn
Mime
View raw message