DataStreamApi 之watermark和开窗设置

曾经终败给现在 2023-10-06 19:52 19阅读 0赞

#### watermark 设置的方式 ####

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            DataStreamSource<Snapshot> dataStreamSource = env.addSource(new SourceFormClickhouse2());
      // 设置watermark
      WatermarkStrategy<Snapshot> snapshotWatermarkStrategy =
                    WatermarkStrategy.<Snapshot>forBoundedOutOfOrderness(Duration.ofSeconds(10)).
                            withTimestampAssigner((s, timestamp) -> DateUtil.dateToStamp(s.getDateTime()));
    	// 引入watermark
     SingleOutputStreamOperator<Snapshot> streamOperator = dataStreamSource.assignTimestampsAndWatermarks(snapshotWatermarkStrategy);

#### 开窗 window (key 分组) ####

多个字段想分组，如何设置。  
之前版本 有 keyBy(“field1,field2,field3”) 这种方法已经废弃。

###### 1、方式一 ######

使用元组来实现多个字段keyBy

streamOperator.keyBy(snapshot -> Tuple2.of(snapshot.getHsSecurityId(), snapshot.getSecurityId()))
                    .window(TumblingEventTimeWindows.of(Time.seconds(15)))
                    .max("highPx")
                    .print();

###### 2、方式二 ######

这种写法性能差些，也可以实现。

streamOperator.keyBy(snapshot -> snapshot.getSecurityId())
                    .keyBy(s->s.getHsSecurityId())
                    .window(TumblingEventTimeWindows.of(Time.seconds(15)))
                    .max("highPx")
                    .print();

###### 3、方式三 ######

使用对象构造

streamOperator.keyBy(snapshot -> new FieldSorting(snapshot.getSecurityId(),snapshot.getSecurityId()))
                    .window(TumblingEventTimeWindows.of(Time.seconds(15)))
                    .max("highPx")
                    .print();

发表评论取消回复

表情：

评论列表（有 0 条评论，19人围观）

还没有评论，来说两句吧...

相关阅读

相关 Spark之开窗函数

一.简介　　开窗函数row\_number()是按照某个字段分组，然后取另外一个字段排序的前几个值的函数，相当于分组topN。如果SQL语句里面使用了开窗函数，那么...

痛定思痛。/ 2024年04月20日 09:37/ 0 赞/ 84 阅读

相关四、Flink DataStreamAPI

*目录** DataStream Transformation map FlatMap Filter KeyBy Reduce Fold 聚合 Wind...

浅浅的花香味﹌/ 2024年04月18日 23:08/ 0 赞/ 69 阅读

相关 C#窗体应用设置多窗体和设置启动窗体

场景 VS中使用C\开发一个登录窗体应用程序： [https://blog.csdn.net/BADAO\_LIUMANG\_QIZHI/article/details

约定不等于承诺〃/ 2023年10月18日 23:53/ 0 赞/ 201 阅读

相关 DataStreamApi 之watermark和开窗设置

watermark 设置的方式 StreamExecutionEnvironment env = StreamExecutionEnvironment.getEx

曾经终败给现在/ 2023年10月06日 19:52/ 0 赞/ 20 阅读

相关 Flink之DataStreamApi编程实战

一、Flink DataStreamApi作用和使用场景 Flink中的DataStream程序是实现在数据流上的transformation(如filtering，up

古城微笑少年丶/ 2023年10月06日 18:09/ 0 赞/ 39 阅读

相关数据流—DataStreamAPI

Hello Flink ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly

以你之姓@/ 2023年10月04日 16:21/ 0 赞/ 17 阅读

相关 mysqlover开窗函数

这句select row\_number() over (order by ProductID asc) as RowIndex, 不明白呵呵~老实说，要是不是看了你的

超、凢脫俗/ 2023年09月25日 10:17/ 0 赞/ 148 阅读

相关 over开窗函数

优秀的示例：本贴结合写下自己的理解 [https://www.cnblogs.com/lanzi/archive/2010/10/26/1861338.html][https

小鱼儿/ 2023年06月10日 04:28/ 0 赞/ 23 阅读

相关 SparkSQL开窗函数

row\_number（）开窗函数其实就是给每个分组的数据，按照其排序的顺序，打上一个分组内的行号比如说，有一个分组date = 20151001,里面有三条数据，11

冷不防/ 2022年05月18日 04:20/ 0 赞/ 353 阅读

相关开窗函数

·格式函数名(列) OVER(选项) ·作用一般的聚集函数返回值是单值，而开窗函数返回值是多值，可能是一列的形式 ·示例 select S,A

川长思鸟来/ 2021年12月17日 03:33/ 0 赞/ 350 阅读