devbox
酷酷是懒虫
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink nc -l -p 监听端口测试

作者：酷酷是懒虫 | 2024-06-24 05:29:23

赞

踩

Flink nc -l -p 监听端口测试

1、9999端口未占用

netstat -apn|grep 9999

2、消息发送端

nc -l -k -p 9999
  1. {"user":"ming","url":"www.baidu1.com", "timestamp":1200L, "score":1}
  2. {"user":"xiaohu","url":"www.baidu5.com","timestamp":1267L, "score":10}
  3. {"user":"ming","url":"www.baidu7.com","timestamp":4200L, "score":9}
  4. {"user":"xiaohu","url":"www.baidu8.com","timestamp":5500L, "score":90}
  5. {"user":"Biu","url":"www.baidu8.com","timestamp":5500L, "score":1000}
  6.  
  7. {"user":"ming","url":"www.baidu1.com", "timestamp":1717171200000, "score":1}
  8. {"user":"xiaohu","url":"www.baidu5.com","timestamp":1717171202000, "score":10}
  9. {"user":"ming","url":"www.baidu7.com","timestamp":1717171260000, "score":9}
  10. {"user":"xiaohu","url":"www.baidu8.com","timestamp":1717264860000, "score":90}
  11. {"user":"Biu","url":"www.baidu8.com","timestamp":1718780790000, "score":1000}

3、运行

周期性水位线

  1. import com.alibaba.fastjson2.JSONObject;
  2. import org.apache.flink.api.common.eventtime.*;
  3. import org.apache.flink.api.common.functions.MapFunction;
  4. import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
  5. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  6. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8.  
  9. import java.sql.Timestamp;
  10. import java.util.ArrayList;
  11.  
  12. /**
  13.  * Description: 
  14.  * forMonotonousTimestamps->AscendingTimestampsWatermarks 有序流 -> 自定义断点式水位线（周期延迟时间=0ms）\
  15.  * forBoundedOutOfOrderness->BoundedOutOfOrdernessWatermarks 无序流 -> 自定义周期性水位线
  16.  */
  17. public class FlinkPeriodicWatermarkGeneratorTestJob {
  18.     public static void main(String[] args) throws Exception {
  19.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  20.  
  21. //        ArrayList<Event> list = new ArrayList<>();
  22. //        list.add(new Event("ming","www.baidu1.com",1200L));
  23. //        list.add(new Event("xiaohu","www.baidu5.com",1267L));
  24. //        list.add(new Event("ming","www.baidu7.com",4200L));
  25. //        list.add(new Event("xiaohu","www.baidu8.com",5500L));
  26. //
  27. //        DataStreamSource<Event> ds = env.fromCollection(list, BasicTypeInfo.of(Event.class));
  28.  
  29.         DataStreamSource<String> dss = env.socketTextStream("test002", 9999);
  30.         SingleOutputStreamOperator<Event> ds = dss.map(new MapFunction<String, Event>() {
  31.             @Override
  32.             public Event map(String value) throws Exception {
  33.                 Event event = new Event();
  34.                 event.toEvent(value);
  35.                 return event;
  36.             }
  37.         });
  38. //        ds.print();
  39.         SingleOutputStreamOperator<Event> watermarks = ds
  40.                 // AscendingTimestampsWatermarks 有序流 查看源码，实际上是延迟时间=0ms的乱序流
  41. //                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
  42.                 // BoundedOutOfOrdernessWatermarks 无序流 5ms固定延迟时间/表示最大乱序程度 处理乱序流数据
  43.                 .assignTimestampsAndWatermarks(new WatermarkStrategy<Event>() {
  44.                     @Override
  45.                     public TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
  46.                         return new SerializableTimestampAssigner<Event>() {
  47.                             @Override
  48.                             public long extractTimestamp(Event element, long recordTimestamp) {
  49.                                 return element.getTimestamp();
  50.                             }
  51.                         };
  52.                     }
  53.  
  54.                     @Override
  55.                     public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
  56.                         return new WatermarkGenerator<Event>() {
  57.                             private Long delayTime = 5000L; // 延迟时间
  58.  
  59.                             private Long maxTs = Long.MIN_VALUE + delayTime + 1L;
  60.  
  61.                             @Override
  62.                             public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {
  63.                                 // 每来一条数据就调用一次
  64.                                 maxTs = Math.max(event.timestamp, maxTs);// 更新最大时间戳
  65.                             }
  66.  
  67.                             @Override
  68.                             public void onPeriodicEmit(WatermarkOutput output) {
  69.                                 // 发射水位线，默认 200ms 调用一次 可以使用 env.getConfig().setAutoWatermarkInterval(60 * 1000L); 调整周期时间 flink时间窗口(左开,右闭]
  70.                                 output.emitWatermark(new Watermark(maxTs - delayTime - 1L));
  71.                             }
  72.                         };
  73.                     }
  74.                 });
  75.  
  76.         ds.print();
  77.  
  78.         env.setParallelism(1);
  79.         env.execute();
  80.     }
  81.  
  82.     public static class Event{
  83.         String user;
  84.         String url;
  85.         Long timestamp;
  86.  
  87.         public Event(){
  88.         }
  89.         public Event(String user, String url, Long timestamp) {
  90.             this.user = user;
  91.             this.url = url;
  92.             this.timestamp = timestamp;
  93.         }
  94.  
  95.         public String getUser() {
  96.             return user;
  97.         }
  98.  
  99.         public String getUrl() {
  100.             return url;
  101.         }
  102.  
  103.         public Long getTimestamp() {
  104.             return timestamp;
  105.         }
  106.  
  107.         @Override
  108.         public String toString() {
  109.             return "Event{" +
  110.                     "user='" + user + '\'' +
  111.                     ", url='" + url + '\'' +
  112.                     ", timestamp=" + new Timestamp(timestamp) +
  113.                     '}';
  114.         }
  115.  
  116.         public void toEvent(String val){
  117.             JSONObject js = JSONObject.parseObject(val);
  118.             this.user = js.getString("user");
  119.             this.url = js.getString("url");
  120.             this.timestamp = js.getLong("timestamp");
  121.         }
  122.     }
  123. }

断点式水位线

  1. import com.alibaba.fastjson2.JSONObject;
  2. import org.apache.flink.api.common.eventtime.*;
  3. import org.apache.flink.api.common.functions.MapFunction;
  4. import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
  5. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  6. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8.  
  9. import java.sql.Timestamp;
  10. import java.util.ArrayList;
  11.  
  12. /**
  13.  * Description: 
  14.  * forMonotonousTimestamps->AscendingTimestampsWatermarks 有序流 -> 自定义断点式水位线（周期延迟时间=0ms）\
  15.  * forBoundedOutOfOrderness->BoundedOutOfOrdernessWatermarks 无序流 -> 自定义周期性水位线
  16.  */
  17. public class FlinkPunctuatedWatermarkGeneratorTestJob {
  18.     public static void main(String[] args) throws Exception {
  19.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  20.  
  21.         DataStreamSource<String> dss = env.socketTextStream("test002", 9999);
  22.         SingleOutputStreamOperator<Event> ds = dss.map(new MapFunction<String, Event>() {
  23.             @Override
  24.             public Event map(String value) throws Exception {
  25.                 Event event = new Event();
  26.                 event.toEvent(value);
  27.                 return event;
  28.             }
  29.         });
  30. //        ds.print();
  31.         SingleOutputStreamOperator<Event> watermarks = ds
  32.                 // AscendingTimestampsWatermarks 有序流 查看源码，实际上是延迟时间=0ms的乱序流
  33. //                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
  34.                 // BoundedOutOfOrdernessWatermarks 无序流 5ms固定延迟时间/表示最大乱序程度 处理乱序流数据
  35.                 .assignTimestampsAndWatermarks(new WatermarkStrategy<Event>() {
  36.                     @Override
  37.                     public TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
  38.                         return new SerializableTimestampAssigner<Event>() {
  39.                             @Override
  40.                             public long extractTimestamp(Event element, long recordTimestamp) {
  41.                                 return element.getTimestamp();
  42.                             }
  43.                         };
  44.                     }
  45.  
  46.                     @Override
  47.                     public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
  48.                         return new WatermarkGenerator<Event>() {
  49.                             @Override
  50.                             public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {
  51.                                 // 只有在遇到特定的 itemId 时，才发出水位线
  52.                                 if (event.getUser().equals("Biu")) {
  53.                                     output.emitWatermark(new Watermark(event.getTimestamp() - 1));
  54.                                 }
  55.                             }
  56.  
  57.                             @Override
  58.                             public void onPeriodicEmit(WatermarkOutput output) {
  59.                                 // 不需要做任何事情，因为我们在 onEvent 方法中发射了水位线
  60.                             }
  61.                         };
  62.                     }
  63.                 });
  64.  
  65.         ds.print();
  66.  
  67.         env.setParallelism(1);
  68.         env.execute();
  69.     }
  70.  
  71.     public static class Event{
  72.         String user;
  73.         String url;
  74.         Long timestamp;
  75.  
  76.         public Event(){
  77.         }
  78.         public Event(String user, String url, Long timestamp) {
  79.             this.user = user;
  80.             this.url = url;
  81.             this.timestamp = timestamp;
  82.         }
  83.  
  84.         public String getUser() {
  85.             return user;
  86.         }
  87.  
  88.         public String getUrl() {
  89.             return url;
  90.         }
  91.  
  92.         public Long getTimestamp() {
  93.             return timestamp;
  94.         }
  95.  
  96.         @Override
  97.         public String toString() {
  98.             return "Event{" +
  99.                     "user='" + user + '\'' +
  100.                     ", url='" + url + '\'' +
  101.                     ", timestamp=" + new Timestamp(timestamp) +
  102.                     '}';
  103.         }
  104.  
  105.         public void toEvent(String val){
  106.             JSONObject js = JSONObject.parseObject(val);
  107.             this.user = js.getString("user");
  108.             this.url = js.getString("url");
  109.             this.timestamp = js.getLong("timestamp");
  110.         }
  111.     }
  112. }
  113.  
  114.

4、打印

  1. 3> Event{user='ming', url='www.baidu1.com', timestamp=1970-01-01 08:00:01.2}
  2. 4> Event{user='xiaohu', url='www.baidu5.com', timestamp=1970-01-01 08:00:01.267}
  3. 5> Event{user='ming', url='www.baidu7.com', timestamp=1970-01-01 08:00:04.2}
  4. 6> Event{user='xiaohu', url='www.baidu8.com', timestamp=1970-01-01 08:00:05.5}

参考：

【Flink】Flink 中的时间和窗口之水位线(Watermark)-CSDN博客

Flink watermark_nc -lp 9999-CSDN博客

NoteWarehouse/05_BigData/09_Flink(1).md at main · FGL12321/NoteWarehouse · GitHub

声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号