Flink的简单学习二

一 Flink的核心组件

1.1 client

1.将数据流程图DataFlow发送给JobManager。

1.2 JobManager

1.收集client的DataFlow图，将图分解成一个个的task任务，并返回状态更新数据给client

2.JobManager负责作业调度，收集TaskManager的Heartbeat和统计信息。

1.3 TaskManager

1.将每一个task任务放到一个TaskSlot槽中

2.TaskManager 之间以流的形式进行数据的传输。

二 Flink的集群搭建

2.1 独立集群

2.1.1 上传解压配置环境变量

1.解压 tar -zxvf flink-1.15.2-bin-scala_2.12.tgz -C ../

2.配置环境变量

# 配置环境变量
vim /etc/profileexport FLINK_HOME=/usr/local/soft/flink-1.15.2
export PATH=$PATH:$FLINK_HOME/binsource /etc/profile

2.1.2 修改配置文件

1.修改flink-conf.yaml

jobmanager.rpc.address: master
jobmanager.bind-host: 0.0.0.0
taskmanager.bind-host: 0.0.0.0
taskmanager.host: localhost # noe1和node2需要单独修改
taskmanager.numberOfTaskSlots: 4
rest.address: master
rest.bind-address: 0.0.0.0

2.修改masters

master:8081

3.修改workers

node1
node2

2.1.3 同步到所有节点

1.同步

scp -r flink-1.15.2 node1:`pwd`
scp -r flink-1.15.2 node2:`pwd`

2.修改子节点的flink-conf.yaml文件中的taskmanager.host

taskmanager.host: node1
taskmanager.host: node2

2.1.4 启动与关闭集群

1.启动

start-cluster.sh

2.看是否成功，打开web ui界面

http://master:8081

3.关闭集群

stop-cluster.sh

2.1.5 提交任务

1.将代码打包到服务器中提交

1.启动命令

flink run -c com.shujia.flink.core.Demo1StreamWordCount flink-1.0.jar

com.shujia.flink.core.Demo1StreamWordCount：主类名

flink-1.0.jar：jar包名

2.查看web界面

3.查看结果

4.关闭任务

 2.web界面提交任务

1.提交

2.相关配置

2.2 Flink on Yarn

2.2.1 整合

1.在环境变量中配置HADOOP_CLASSSPATH

vim /etc/profileexport HADOOP_CLASSPATH=`hadoop classpath`source /etc/profile

 2.2.2 Application Mode

1、将任务提交到yarn上运行，yarn会为每一个flink地任务启动一个jobmanager和一个或者多个taskmanasger

2、代码main函数不再本地运行，dataFlow不再本地构建，如果代码报错在本地看不到详细地错误日志

 1.启动命令

flink run-application -t yarn-application -c com.shujia.flink.core.Demo1StreamWordCount flink-1.0.jar

flink run-application -t yarn-application -c：任务命令名

com.shujia.flink.core.Demo1StreamWordCount：主类名

flink-1.0.jar：jar包名

2.查看界面

点击这个，直接跳转到Flink的web界面

2.2.3 Per-Job Cluster Mode

1、将任务提交到yarn上运行，yarn会为每一个flink地任务启动一个jobmanager和一个或者多个taskmanasger

2、代码地main函数在本地启动，在本地构建dataflow，再将dataflow提交给jobmanager,如果代码报错再本地可以烂到部分错误日志

1.启动命令

flink run -t yarn-per-job -c com.shujia.flink.core.Demo1StreamWordCount flink-1.0.jar

flink run -t yarn-per-job -c：命令名

com.shujia.flink.core.Demo1StreamWordCount：主类名

flink-1.0.jar：jar包名

2.界面跟Application Mode一样

2.3.4 Session Mode

1、先再yarn中启动一个jobmanager, 不启动taskmanager

2、提交任务地时候再动态申请taskmanager

3、所有使用session模式提交的任务共享同一个jobmanager

4、类似独立集群，只是集群在yarn中启动了，可以动态申请资源

5、一般用于测试

1.先启动会话集群

yarn-session.sh -d

启动过后出现这个，一个是任务编码application_1717379968853_0003

另一个是web界面，复制可以打开

2.提交任务

命令提交：

flink run -t yarn-session -Dyarn.application.id=application_1717379968853_0003 -c com.shujia.flink.core.Demo1StreamWordCount flink-1.0.jar

Dyarn.application.id=application_1717379968853_0003：这个是启动会话集群给的

com.shujia.flink.core.Demo1StreamWordCount：主类名

flink-1.0.jar：jar包名

web界面提交：跟Application Mode的web提交一模一样

三 并行度

3.1 设置并行度

3.1.1 代码中设置

1.代码中不设置，默认的并行度数量是配置文件里面的

2.代码中配置

env.setParallelism(2)

3.1.2 提交任务中设置

1.加一个参数 -p 并行度数量

例如：

flink run -t yarn-session -p 3 -Dyarn.application.id=application_1717379968853_0003 -c com.shujia.flink.core.Demo1StreamWordCount flink-1.0.jar

2.或者在ui界面中设置

3.1.3 配置文件中设置

1.这个一般不用

在flink-conf.yaml修改配置

3.1.4 每一个算子单独设置

在代码中使用算子时候后面可以设置并行度，但是这种不用

3.1.4 优先级

代码>提交任务中配置>配置文件

3.2 共享资源

 1、flink需要资源的数量和task数量无关

 2、一个并行度对应一个资源（slot）

 3、上游task的下游task共享同一个资源

3.3 并行度设置原则

1.实时计算的任务并行度取决于数据的吞吐量

 2、聚合计算（有shuffle）的代码一个并行度大概一秒可以处理10000条数据左右

 3、非聚合计算是，一个并行度大概一秒可以处理10万条左右

四 事件时间

4.1 event time

数据产生的时间，数据中有一个时间字段，使用数据的时间字段触发计算，代替真实的时间，可以反应数据真实发生的顺序，计算更有意义

4.1.1 数据时间无乱序

1.解析数据，分析哪个数据是数据时间

2.指定时间字段

forMonotonousTimestamps()：单调递增。数据时间只能是往上增的

tsDS.assignTimestampsAndWatermarks(WatermarkStrategy
//指定水位线生产策略，水位线等于最新一条数据的时间戳，如果数据乱序可能会丢失数据
.<Tuple2<String, Long>>forMonotonousTimestamps()
//指定时间字段
.withTimestampAssigner((event, ts) -> event.f1));

2.完整代码如下 

package com.shujia.flink.core;import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;import java.time.Duration;public class Demo5EventTime {public static void main(String[] args)throws Exception {/** 事件时间：数据中有一个时间字段，使用数据的时间字段触发计算，代替真实的时间，可以反应数据真实发生的顺序，计算更有意义*//*java,1717395300000java,1717395301000java,1717395302000java,1717395303000java,1717395304000java,1717395305000*/StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStream<String> linesDS = env.socketTextStream("master", 8888);//解析数据DataStream<Tuple2<String, Long>> tsDS = linesDS.map(line -> {String[] split = line.split(",");String word = split[0];long ts = Long.parseLong(split[1]);return Tuple2.of(word, ts);}, Types.TUPLE(Types.STRING, Types.LONG));/** 指定时间字段和水位线生成策略*/DataStream<Tuple2<String, Long>> assDS = tsDS.assignTimestampsAndWatermarks(WatermarkStrategy//指定水位线生产策略，水位线等于最新一条数据的时间戳，如果数据乱序可能会丢失数据.<Tuple2<String, Long>>forMonotonousTimestamps()//指定时间字段.withTimestampAssigner((event, ts) -> event.f1));/**每隔5秒统计单词的数量*/DataStream<Tuple2<String, Integer>> kvDS = assDS.map(kv -> Tuple2.of(kv.f0, 1), Types.TUPLE(Types.STRING, Types.INT));KeyedStream<Tuple2<String, Integer>, String> keyByDS = kvDS.keyBy(kv -> kv.f0);//TumblingEventTimeWindows:滚动的事件时间窗口WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowDS = keyByDS.window(TumblingEventTimeWindows.of(Time.seconds(5)));windowDS.sum(1).print();env.execute();}
}

3.结果分析

上面代码是以5秒作为一个滚动的事件时间窗口。不包括第五秒，左闭右开。

窗口的触发条件：水位线大于等于窗口的结束时间；窗口内有数据

水位线：等于最新一条数据的时间戳

比如说0-5-10-15-20.0-5是一个窗口，5-10是一个窗口，且窗口里面有数据才能被计算，如果这个窗口里面出现了不存在这个时间的事件，则不会被处理

输入的事件时间是乱序的，他丢失第四次输出的。

 4.1.2 数据时间乱序

1.水位线前移，使用forBoundedOutOfOrderness里面传入前移的时间

tsDS.assignTimestampsAndWatermarks(WatermarkStrategy
//水位线前移时间（数据最大乱序时间）
.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
//指定时间字段
.withTimestampAssigner((event, ts) -> event.f1));

2.完整代码

package com.shujia.flink.core;import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;import java.time.Duration;public class Demo5EventTime {public static void main(String[] args)throws Exception {/** 事件时间：数据中有一个时间字段，使用数据的时间字段触发计算，代替真实的时间，可以反应数据真实发生的顺序，计算更有意义*/StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();/*java,1717395301000java,1717395302000java,1717395303000java,1717395304000java,1717395305000java,1717395307000java,1717395308000java,1717395311000java,1717395313000java,1717395315000*/env.setParallelism(1);DataStream<String> linesDS = env.socketTextStream("master", 8888);//解析数据DataStream<Tuple2<String, Long>> tsDS = linesDS.map(line -> {String[] split = line.split(",");String word = split[0];long ts = Long.parseLong(split[1]);return Tuple2.of(word, ts);}, Types.TUPLE(Types.STRING, Types.LONG));/** 指定时间字段和水位线生成策略*/DataStream<Tuple2<String, Long>> assDS = tsDS.assignTimestampsAndWatermarks(WatermarkStrategy//指定水位线生产策略，水位线等于最新一条数据的时间戳，如果数据乱序可能会丢失数据
//                                .<Tuple2<String, Long>>forMonotonousTimestamps()//水位线前移时间（数据最大乱序时间）.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))//指定时间字段.withTimestampAssigner((event, ts) -> event.f1));/**每隔5秒统计单词的数量*/DataStream<Tuple2<String, Integer>> kvDS = assDS.map(kv -> Tuple2.of(kv.f0, 1), Types.TUPLE(Types.STRING, Types.INT));KeyedStream<Tuple2<String, Integer>, String> keyByDS = kvDS.keyBy(kv -> kv.f0);//TumblingEventTimeWindows:滚动的事件时间窗口WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowDS = keyByDS.window(TumblingEventTimeWindows.of(Time.seconds(5)));windowDS.sum(1).print();env.execute();}
}

3.结果分析

我输入的如图所示，我代码设置了水位线前移5秒中，所以触发时间是10秒才触发任务，0-10秒里有4个0-5里面的数据，所以输出了4.为什么14000没有输出，因为14-5=9，他还没有到下一阶段的水位线。我再输出了16秒的，他就有结果了。

4.1.3 水位线对齐

1.当上游有多个task时，下游task会取上游task水位线的最小值，如果数据量小。水位线就很难对齐，窗口就不会触发计算。故要设置并行度，提前把task设定好。

2.如果不设置并行度，可能要输出很多事件才能触发计算。

4.2 processing time

1.处理时间：真实时间

2.这个代码是设置了滚动的处理时间窗口吗，每现实时间5秒中处理一下数据

package com.shujia.flink.core;import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;public class Demo4ProcTime {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStream<String> wordsDS = env.socketTextStream("master", 8888);//转换成kvDataStream<Tuple2<String, Integer>> kvDS = wordsDS.map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT));//按照单词分组KeyedStream<Tuple2<String, Integer>, String> keyByDS = kvDS.keyBy(kv -> kv.f0);//划分窗口//TumblingProcessingTimeWindows:滚动的处理时间窗口WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowDS = keyByDS.window(TumblingProcessingTimeWindows.of(Time.seconds(5)));//统计单词的数量DataStream<Tuple2<String, Integer>> countDS = windowDS.sum(1);countDS.print();env.execute();}
}

五 窗口

5.1 time window

1.时间窗口有四种：

SlidingEventTimeWindows：滑动的事件时间窗口

SlidingProcessingTimeWindows: 滑动的处理时间窗口 

TumblingEventTimeWindows:滚动的事件时间窗口

TumblingProcessingTimeWindows:滚动的处理时间窗口

2.滑动事件需要设置2个时间，一个设置窗口的大小，另一个是滚动的时间

package com.shujia.flink.window;import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;import java.time.Duration;public class Demo1TimeWindow {public static void main(String[] args)throws Exception {/** 事件时间：数据中有一个时间字段，使用数据的时间字段触发计算，代替真实的时间，可以反应数据真实发生的顺序，计算更有意义*/StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();/*java,1717395301000java,1717395302000java,1717395303000java,1717395304000java,1717395305000java,1717395307000java,1717395308000java,1717395311000java,1717395313000java,1717395315000*//**水位线对齐* 1、当上游有多个task时，下游task会取上游task水位线的最小值，如果数据量小。水位线就很难对齐，窗口就不会触发计算*/env.setParallelism(1);DataStream<String> linesDS = env.socketTextStream("master", 8888);//解析数据DataStream<Tuple2<String, Long>> tsDS = linesDS.map(line -> {String[] split = line.split(",");String word = split[0];long ts = Long.parseLong(split[1]);return Tuple2.of(word, ts);}, Types.TUPLE(Types.STRING, Types.LONG));/** 指定时间字段和水位线生成策略*/DataStream<Tuple2<String, Long>> assDS = tsDS.assignTimestampsAndWatermarks(WatermarkStrategy//指定水位线生产策略，水位线等于最新一条数据的时间戳，如果数据乱序可能会丢失数据
//                                .<Tuple2<String, Long>>forMonotonousTimestamps()//水位线前移时间（数据最大乱序时间）.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))//指定时间字段.withTimestampAssigner((event, ts) -> event.f1));/**每隔5秒统计单词的数量*/DataStream<Tuple2<String, Integer>> kvDS = assDS.map(kv -> Tuple2.of(kv.f0, 1), Types.TUPLE(Types.STRING, Types.INT));KeyedStream<Tuple2<String, Integer>, String> keyByDS = kvDS.keyBy(kv -> kv.f0);/** SlidingEventTimeWindows：滑动的事件时间窗口* SlidingProcessingTimeWindows: 滑动的处理时间窗口* TumblingEventTimeWindows:滚动的事件时间窗口* TumblingProcessingTimeWindows:滚动的处理时间窗口* 滑动的时间窗口需要设置两个时间，第一个是窗口的大小，第二个是记录的时间，* 比如说(15,5),这是每5秒计算最近15秒内的数据*/WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowDS = keyByDS.window(SlidingEventTimeWindows.of(Time.seconds(15),Time.seconds(5)));windowDS.sum(1).print();env.execute();}
}

这个代码用的是滑动的事件时间窗口，我设置了每5秒钟计算最近15秒内的数据

5.2 count time

1.滚动的统计窗口：每个key隔多少数据计算一次

package com.shujia.flink.window;import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;public class Demo2CountWindow {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStream<String> linesDS = env.socketTextStream("master", 8888);DataStream<Tuple2<String, Integer>> kvDS = linesDS.map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT));KeyedStream<Tuple2<String, Integer>, String> keyByDS = kvDS.keyBy(kv -> kv.f0);/** 统计窗口* countWindow(10)：滚动的统计窗口， 每个key每隔10条数据计算一次* countWindow(10, 2): 滑动的统计窗口，每隔两条数据计算最近10条数据*/WindowedStream<Tuple2<String, Integer>, String, GlobalWindow> countWindowDS = keyByDS.countWindow(10, 2);countWindowDS.sum(1).print();env.execute();}
}

 2.滑动的统计窗口：每隔多少数据计算最近的多少条数据

package com.shujia.flink.window;import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;public class Demo2CountWindow {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStream<String> linesDS = env.socketTextStream("master", 8888);DataStream<Tuple2<String, Integer>> mapDS = linesDS.map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT));KeyedStream<Tuple2<String, Integer>, String> keyBy = mapDS.keyBy(kv -> kv.f0);WindowedStream<Tuple2<String, Integer>, String, GlobalWindow> countWindow = keyBy.countWindow(10,2);countWindow.sum(1).print();env.execute();}
}

5.3 session time

1.处理时间的会话窗口ProcessingTimeSessionWindows：对一个key，10秒内没有下一步数据开始计算。比如说我输入了 a*7次，然后等10秒输出结果是(a,7)。我再输入a*6次加一个aa，那么输出结果是(aa,1)与(a,6).

package com.shujia.flink.window;import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;public class Demo3SessionWindow {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStream<String> linesDS = env.socketTextStream("master", 8888);DataStream<Tuple2<String, Integer>> mapDS = linesDS.map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT));KeyedStream<Tuple2<String, Integer>, String> keyBy = mapDS.keyBy(kv -> kv.f0);WindowedStream<Tuple2<String, Integer>, String, TimeWindow> window = keyBy.window(ProcessingTimeSessionWindows.withGap(Time.seconds(10)));window.sum(1).print();env.execute();}
}

2.事件时间的会话窗口EventTimeSessionWindows：根据数据的时间，对应同一个key，10秒内没有下一步数据开始计算

这个不常用

package com.shujia.flink.window;import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;import java.time.Duration;public class Demo4EventTimeSessionWindow {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();/*
java,1685433130000
java,1685433131000
java,1685433132000
java,1685433134000
java,1685433135000
java,1685433137000
java,1685433139000
java,1685433149000
java,1685433155000
java,1685433170000*/env.setParallelism(1);DataStream<String> linesDS = env.socketTextStream("master", 8888);//解析数据DataStream<Tuple2<String, Long>> tsDS = linesDS.map(line -> {String[] split = line.split(",");String word = split[0];long ts = Long.parseLong(split[1]);return Tuple2.of(word, ts);}, Types.TUPLE(Types.STRING, Types.LONG));/** 指定时间字段和水位线生成策略*/DataStream<Tuple2<String, Long>> assDS = tsDS.assignTimestampsAndWatermarks(WatermarkStrategy//水位线前移时间（数据最大乱序时间）.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))//指定时间字段.withTimestampAssigner((event, ts) -> event.f1));/**每隔5秒统计单词的数量*/DataStream<Tuple2<String, Integer>> kvDS = assDS.map(kv -> Tuple2.of(kv.f0, 1), Types.TUPLE(Types.STRING, Types.INT));KeyedStream<Tuple2<String, Integer>, String> keyByDS = kvDS.keyBy(kv -> kv.f0);/** EventTimeSessionWindows:事件时间的会话窗口*/WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowDS = keyByDS.window(EventTimeSessionWindows.withGap(Time.seconds(10)));windowDS.sum(1).print();env.execute();        }
}

5.4 process与窗口结合

1.设置了窗口过后的DS后面用process算子，他里面传入的是实现ProcessWindowFunction中的抽象方法process的对象，这个抽象类里面传的是4个参数(IN, OUT, KEY, W)，输入的类型，输出的类型，key的类型，以及窗口类型。窗口类型是三大窗口的其中之一。

2.process方法里面，第一个参数是key，第二个参数是flink的环境连接对象。第三个参数是kv的键值对，第四个参数是发送的对象

代码如下

package com.shujia.flink.window;import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;public class Demo5WindowProcess {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStream<String> linesDS = env.socketTextStream("master", 8888);SingleOutputStreamOperator<Tuple2<String, Integer>> kvDS = linesDS.map(line -> {String[] lines = line.split(",");String clazz = lines[4];int age = Integer.parseInt(lines[2]);return Tuple2.of(clazz, age);}, Types.TUPLE(Types.STRING, Types.INT));KeyedStream<Tuple2<String, Integer>, String> keyBy = kvDS.keyBy(kv -> kv.f0);WindowedStream<Tuple2<String, Integer>, String, TimeWindow> window = keyBy.window(TumblingProcessingTimeWindows.of(Time.seconds(5)));DataStream<Tuple2<String, Double>> process = window.process(new ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Double>, String, TimeWindow>() {@Overridepublic void process(String clazz,ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Double>, String, TimeWindow>.Context context,Iterable<Tuple2<String, Integer>> elements,Collector<Tuple2<String, Double>> out) throws Exception {double sum_age = 0;int num = 0;for (Tuple2<String, Integer> element : elements) {sum_age += element.f1;num++;}double avg_age = sum_age / num;out.collect(Tuple2.of(clazz, avg_age));}});process.print();env.execute();}
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/web/21718.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

三、基于图像分类预训练编码及图神经网络的预测模型 【框图+源码】

三、基于图像分类预训练编码及图神经网络的预测模型 【框图+源码】

背景&#xff1a; 抽时间补充&#xff0c;先挖个坑。 一、模型结构 二、源码
阅读更多...
美团一面：什么是CAS？有什么优缺点？我说你说的是AtomicInteger吗？

美团一面：什么是CAS？有什么优缺点？我说你说的是AtomicInteger吗？

引言 传统的并发控制手段&#xff0c;如使用synchronized关键字或者ReentrantLock等互斥锁机制&#xff0c;虽然能够有效防止资源的竞争冲突&#xff0c;但也可能带来额外的性能开销&#xff0c;如上下文切换、锁竞争导致的线程阻塞等。而此时就出现了一种乐观锁的策略&#x…
阅读更多...
企业数字化转型的测度难题：基于大语言模型的新方法与新发现

企业数字化转型的测度难题：基于大语言模型的新方法与新发现

《经济研究》新文章《企业数字化转型的测度难题&#xff1a;基于大语言模型的新方法与新发现》运用机器学习和大语言模型构造一套新的企业数字化转型指标。理论分析和数据交叉验证均表明&#xff0c;构建的指标相对已有方法更准确&#xff1a; 1.第一步&#xff1a;选择“管理…
阅读更多...
17.Redis之主从复制

17.Redis之主从复制

1.主从复制是怎么回事&#xff1f; 分布式系统, 涉及到一个非常关键的问题: 单点问题 单点问题&#xff1a;如果某个服务器程序, 只有一个节点(只搞一个物理服务器, 来部署这个服务器程序) 1.可用性问题,如果这个机器挂了,意味着服务就中断了~ 2.性能/支持的并发量也是比较有限…
阅读更多...
【HarmonyOS】鸿蒙系统中应用权限等级介绍、定义、申请授权讲解

【HarmonyOS】鸿蒙系统中应用权限等级介绍、定义、申请授权讲解

【HarmonyOS】鸿蒙系统中应用权限等级介绍、定义、申请授权讲解 针对权限等级&#xff0c;相对于主体来说&#xff0c;会有不同的细分概念。 一、权限APL等级&#xff1a; 首先在鸿蒙系统中&#xff0c;对于权限本身&#xff0c;分为三个等级&#xff1a;normal&#xff0c;s…
阅读更多...
SQL面试问题集

SQL面试问题集

目录 Q.左连接和右连接的区别 Q.union 和 union all的区别 1、取结果的交集 2、获取结果后的操作 Q.熟悉开窗函数吗&#xff1f;讲一下row_number和dense_rank的区别。 Q.hive行转列怎么操作的 Q.要求手写的题主要考了聚合函数和窗口函数&#xff0c;row_number()&#…
阅读更多...
同一个tomcat不同端口运行不同项目

同一个tomcat不同端口运行不同项目

第一步&#xff1a;修改 server.xml 文件 修改 tomcat 安装目录下 conf/server.xml 文件&#xff0c;需要几个端口就添加几个 Service 节点。 配置 2 个端口&#xff1a;9131 和 9133&#xff0c;于是增加两个 Service 节点。 每个 Service 节点的 name 属性值要设置不同的值…
阅读更多...
【MATLAB】雷达信号处理程序源码 雷达系统仿真代码 matlab SAR

【MATLAB】雷达信号处理程序源码 雷达系统仿真代码 matlab SAR

【MATLAB】雷达信号处理程序源码 雷达系统仿真代码 matlab SAR 包含以下所有源码,内容如下&#xff1a;&#xff1a; 1、 MATGPR R3探地雷达数据处理 MATLAB 程序 2、 python 雷达图像识别 3、 SAR 雷达回波仿真 matlab 4、 SAR 雷达影像处理源码 5、 STFT 处理 IPIX 雷达…
阅读更多...
数据分析常用模型合集（三）同期群、逻辑树、假设检验等

数据分析常用模型合集（三）同期群、逻辑树、假设检验等

前面两篇文章&#xff0c;我们将比较大、较为系统的分析方法作了一个介绍&#xff0c;本文是最后一篇&#xff0c;将剩余的一些讲一讲。 数据分析常用模型合集&#xff08;二&#xff09;RARRA模型、RFM模型-CSDN博客 剩下的一些模型&#xff0c;其实不应叫做模型&#xff0c;…
阅读更多...
qt+ffmpeg 实现音视频播放（四）之音视频同步

qt+ffmpeg 实现音视频播放（四）之音视频同步

在处理音视频数据时&#xff0c;解码音频的数据往往会比解码视频的数据比较慢&#xff0c;所以我们在播放音视频时&#xff0c;音频和视频的数据会出现渐渐对不上的情况。尤其在播放时间越长的时候&#xff0c;这种对不上的现象越明显。 为了解决这一问题&#xff0c;人们想出…
阅读更多...
在windows操作系统上安装MariaDB

在windows操作系统上安装MariaDB

最近收到关于数据库在哪里看的评论&#xff0c;所以就一不做二不休&#xff0c;把安装数据库的步骤写一篇文章吧。 这篇文章介绍如何在windows上完成MariaDB-10.6.5版本的安装&#xff0c;对应MySQL-8.x版本。 第一步&#xff1a;下载安装包 通过以下网盘链接下载MariaDB-10.6…
阅读更多...
国产信创CPU之飞腾CPU剖析

国产信创CPU之飞腾CPU剖析

CPU&#xff1a;信创根基&#xff0c;国之重器 国产CPU已形成自主架构、x86、ARM三大阵营。自主阵营中&#xff0c;龙芯、申威分别基于MIPS和Alpha推出loong ISA和SW-64。ARM阵营以鲲鹏、飞腾为代表&#xff0c;利用ARM IP授权开发处理器。x86阵营由海光、兆芯等主导&#xff…
阅读更多...
【Linux】操作系统中的文件系统管理：磁盘结构、逻辑存储与文件访问机制

【Linux】操作系统中的文件系统管理：磁盘结构、逻辑存储与文件访问机制

文章目录 前言&#xff1a;1. 磁盘机械结构2. 磁盘物理结构3. 磁盘的逻辑存储3. 1. 文件名呢&#xff1f;3.2 对文件的增删查改与 路径3.3. 文件 4. 软硬链接4.1. 操作观察现象4.2. 软硬链接的原理4.3. 软硬链接的应用场景 总结 前言&#xff1a; 在现代操作系统中&#xff0c…
阅读更多...
基于Springboot+vue实现的汽车服务管理系统

基于Springboot+vue实现的汽车服务管理系统

作者主页&#xff1a;Java码库 主营内容&#xff1a;SpringBoot、Vue、SSM、HLMT、Jsp、PHP、Nodejs、Python、爬虫、数据可视化、小程序、安卓app等设计与开发。 收藏点赞不迷路 关注作者有好处 文末获取源码 技术选型 【后端】&#xff1a;Java 【框架】&#xff1a;spring…
阅读更多...
45.自定义线程池（三）-拒绝策略

45.自定义线程池（三）-拒绝策略

拒绝策略采用函数式接口参数传入&#xff0c;策略模式 FunctionalInterface public interface RejectPolicy<T> {void reject(BlockingQueue<T> queue, T task); } package com.xkj.thread.pool;import com.aspose.words.Run; import lombok.extern.slf4j.Slf4j;…
阅读更多...
SaaS 电商设计 (十一) 那些高并发电商系统的限流方案设计

SaaS 电商设计 (十一) 那些高并发电商系统的限流方案设计

目录 一.什么是限流二.怎么做限流呢2.1 有哪些常见的系统限流算法2.1.1 固定窗口2.1.1 滑动窗口2.1.2 令牌桶2.1.3 漏桶算法 2.2 常见的限流方式2.2.1 单机限流&集群限流2.2.2 前置限流&后置限流 2.3 实际落地是怎么做的2.3.1 流量链路2.3.2 各链路限流2.3.2.1 网关层2…
阅读更多...
重学java 56. Map集合

重学java 56. Map集合

我们要拥有一定成功的信念 —— 24.6.3 一、双列集合的集合框架 HashMap 1.特点: a.key唯一,value可重复 b.无序 c.无索引 d.线程不安全 e.可以存null键,null值 2.数据结构:哈希表 LinkedHashMap&#xff08;继承HashMap&#xff09; 1.特点: a.key唯一,value可重复 b.有序 c.无…
阅读更多...
矩阵连乘问题

矩阵连乘问题

#include<iostream> using namespace std; #define N 7 void MatrixChain(int p[N],int n,int m[N][N],int s[N][N]) {for(int i1;i<n;i)m[i][i]0;for(int r2;r<n;r)//有多少个相乘(规模){for(int i1;i<n-r1;i){int jir-1;m[i][j]m[i][i]m[i1][j]p[i]*p[i1]*p[j…
阅读更多...
小熊家务帮day10- 门户管理

小熊家务帮day10- 门户管理

门户管理 1 门户介绍1.1 介绍1.2 常用技术方案 2 缓存技术方案2.1 需求分析2.1.1 C端用户界面原型2.1.2 缓存需求2.1.3 使用的工具 2.2 项目基础使用2.2.1 项目集成SpringCache2.2.2 测试Cacheable需求Service测试 2.1.3 缓存管理器&#xff08;设置过期时间&#xff09;2.1.4 …
阅读更多...
深入理解序列化：概念、应用与技术

深入理解序列化：概念、应用与技术

在计算机科学中&#xff0c;序列化&#xff08;Serialization&#xff09;是指将数据结构或对象状态转换为可存储或传输的格式的过程。这个过程允许将数据保存到文件、内存缓冲区&#xff0c;或通过网络传输至其他计算机环境&#xff0c;不受原始程序语言的限制。相对地&#x…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023