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.rhkb.cn/news/342613.html

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

相关文章

精选网络安全书单：打造数字世界的钢铁长城！

精选网络安全书单：打造数字世界的钢铁长城！

目录 1.前言 2.书单推荐 2.1. 《内网渗透实战攻略》 2.2. 《Kali Linux高级渗透测试&#xff08;原书第4版&#xff09;》 2.3. 《CTF那些事儿》 2.4. 《权限提升技术&#xff1a;攻防实战与技巧》 2.5. 《数字政府网络安全合规性建设指南&#xff1a;密码应用与数据安全…
阅读更多...
华为鲲鹏应用开发基础： 计算机系统概述(一)

华为鲲鹏应用开发基础： 计算机系统概述(一)

1. 计算机系统演进及分类 1.1 计算机发展的四个阶段 1.2 当前计算机通常分为以下五类： 分类超级计算机大型计算机迷你计算机（服务器）微型计算机工作站特点• 功能最强、运算速度最快、 存储容量最大的计算机 • 多用于国家高科技领域和 尖端技术研究 例如，“神威太湖之光”…
阅读更多...
数字智能数字人直播带货软件系统 实现真人形象的1:1克隆 前后端分离 带完整的安装代码包以及搭建教程

数字智能数字人直播带货软件系统 实现真人形象的1:1克隆 前后端分离 带完整的安装代码包以及搭建教程

系统概述 数字智能数字人直播带货小程序源码系统是一套集人工智能、3D建模、云计算等技术于一体的综合性解决方案。该系统通过深度学习算法&#xff0c;能够实现对真人形象的精准捕捉和1:1克隆&#xff0c;使数字人在直播过程中呈现出与真人无异的表现力。同时&#xff0c;系统…
阅读更多...
使用Kimi月之暗面快速完成学术论文【全流程】

使用Kimi月之暗面快速完成学术论文【全流程】

学境思源&#xff0c;一键生成论文初稿&#xff1a; AcademicIdeas - 学境思源AI论文写作 国内大型互联网公司如阿里、腾讯、360纷纷开始免费提供长文本生成服务&#xff0c;体验了一把国产级的模型Kimi&#xff0c;小编只有一个感觉&#xff1a;国产AI模型只能说越来越牛逼了…
阅读更多...
06- 数组的基础知识详细讲解

06- 数组的基础知识详细讲解

06- 数组的基础知识详细讲解 一、基本概念 一次性定义多个相同类型的变量&#xff0c;并且给它们分配一片连续的内存。 int arr[5];1.1 初始化 只有在定义的时候赋值&#xff0c;才可以称为初始化。数组只有在初始化的时候才可以统一赋值。 以下是一些示例规则&#xff1a; …
阅读更多...
NocoDB开源的智能表格详解-腾讯文档本地替代品

NocoDB开源的智能表格详解-腾讯文档本地替代品

文章目录 一、介绍二、docker-compose部署三、登录NocoDB四、NocoDB手册1. 创建项目2. 收集统计表2.1 添加字段2.2 编辑字段2.3 字段类型2.4 发布表格 3.创建表单3.1 创建表单3.2 分享表单3.3 填写检测单 4.创建看板5.创建画廊 一、介绍 可作为腾讯文档的本地电子表格替代品&a…
阅读更多...
mysql启动出现Error: 2 (No such file or directory)

mysql启动出现Error: 2 (No such file or directory)

查看mydql状态 systemctl status mysqlThe designated data directory /var/lib/mysql/ is unusable 查看mysql日志 tail -f /var/log/mysql/error.logtail: cannot open ‘/var/log/mysql/error.log’ for reading: No such file or directory tail: no files remaining 第…
阅读更多...
小白跟做江科大32单片机之按键控制LED

小白跟做江科大32单片机之按键控制LED

原理部分 1.LED部分使用的是这样的连接方式 2.传感器模块的电路图 滤波电容如果接地&#xff0c;一般用于滤波&#xff0c;在分析电路时就不用考虑。下面这个电路就是看A端和B端哪端的拉力大&#xff0c;就能把电压值对应到相应的电压值 比较器部分 如果A端电压>B端电压&am…
阅读更多...
Android 图表开发开源库 MPAndroidChart 使用总结

Android 图表开发开源库 MPAndroidChart 使用总结

1. 引言 电视项目中需要一个折线图表示节电数据变化情况&#xff0c;类比 H5 来说&#xff0c;Android 中也应该有比较成熟的控件&#xff0c;经过调研后&#xff0c;发现 MPAndroidChart 功能比较强大&#xff0c;网上也有人说可能是目前 Android 开发最好用的一个三方库了&a…
阅读更多...
WPS表格插件方方格子【凑数】功能:选出和等于固定数字的数

WPS表格插件方方格子【凑数】功能:选出和等于固定数字的数

文章目录 后来发现可以下载方方格子插件&#xff0c;使用【凑数】功能https://ffcell.lanzouj.com/iwhfc1kjhayh【凑数】快速【凑数】 导师让沾发票&#xff0c;需要选出若干个数额的发票&#xff0c;使它们的和等于一个指定数。不知道怎么办了&#xff0c;查了一下&#xff0c…
阅读更多...
Python第二语言（四、Python数据容器）

Python第二语言（四、Python数据容器）

目录 一、 数据容器&#xff08;list、tuple、str、map、dict&#xff09; 1. 数据容器概念 2. 列表list&#xff08; [] &#xff09; 2.1 定义方式 2.2 嵌套列表 2.3 list通过获取下标索引获取值 2.4 下标使用概念 2.5 list列表的使用&#xff08;列表的方法&#xff…
阅读更多...
linux nohup命令详解：持久运行命令，无视终端退出

linux nohup命令详解：持久运行命令，无视终端退出

nohup &#xff08;全称为 “no hang up”&#xff09;&#xff0c;用于运行一个命令&#xff0c;使其在你退出 shell 或终端会话后继续运行。 基本语法 nohup command [arg1 ...] [&> output_file] &command 是你想要运行的命令。[arg1 ...] 是该命令的参数。&am…
阅读更多...
快排与归并的算法(非递归版)

快排与归并的算法(非递归版)

一.快排 1.递归法(方法多样) 1>hoare版 注&#xff1a;该方法小编已经在上篇博客中介绍过了&#xff0c;就不在这里过多赘述了&#xff0c;如果有兴趣的小伙伴可以看看小编的上篇博客哦 2>挖坑法 1&#xff09;方法介绍&#xff1a;定义最左边的数据为key&#xff0…
阅读更多...
生信学习入门常见错误可能的原因分类总结和求助指南

生信学习入门常见错误可能的原因分类总结和求助指南

文件或目录找不到 这是常见问题&#xff0c;常见提示有 No such file or directory Error in file(file, “rt”)&#xff1a;无法打开链接 Fatal error: Unable to open file for reading (seq/WT1_1.fq) Fatal error: Unable to read from file (C:Program file/Git/usea…
阅读更多...
OneDrive空间清理及文件历史版本查询

OneDrive空间清理及文件历史版本查询

点击OneDrive图标 点击“在线查看” 点击“设置” 点击“OneDrive设置” 点击“其他设置” 点击“存储标准” 点击“文档” 选择需要操作的文件&#xff0c;点击“历史版本记录” 需要清理空间&#xff0c;可删除历史版本&#xff0c;需要使用历史版本&#xff0c;可还原历史版…
阅读更多...
为什么我们需要在软件本地化过程中使用术语服务？

为什么我们需要在软件本地化过程中使用术语服务？

你知道软件翻译和本地化的术语服务吗&#xff1f;此解决方案涵盖源术语和目标术语的创建、开发和维护。所有术语都存储在具有多个字段的数据库中&#xff0c;包括术语定义、用法示例、上下文和历史记录。这使我们能够正确处理每个术语的创建或更改请求&#xff0c;避免创建重复…
阅读更多...
f1c100s 荔枝派 系统移植

f1c100s 荔枝派 系统移植

一。交叉编译环境配置 1.1下载交叉工具链&#xff1a;https://releases.linaro.org/components/toolchain/binaries/7.2-2017.11/arm-linux-gnueabi/ 1.2解压安装 在home目录下新建 工程目录&#xff1a;mkdir f1c100s_project 将windows下的gcc-linaro-7.2.1-2017.11-x86…
阅读更多...
Spark MLlib 机器学习详解

Spark MLlib 机器学习详解

目录 &#x1f349;引言 &#x1f349;Spark MLlib 简介 &#x1f348; 主要特点 &#x1f348;常见应用场景 &#x1f349;安装与配置 &#x1f349;数据处理与准备 &#x1f348;加载数据 &#x1f348;数据预处理 &#x1f349;分类模型 &#x1f348;逻辑回归 &a…
阅读更多...
C# FTP/SFTP 详解及连接 FTP/SFTP 方式示例汇总

C# FTP/SFTP 详解及连接 FTP/SFTP 方式示例汇总

文章目录 1、FTP/SFTP基础知识FTPSFTP 2、FTP连接示例3、SFTP连接示例4、总结 在软件开发中&#xff0c;文件传输是一个常见的需求。尤其是在不同的服务器之间传输文件时&#xff0c;FTP&#xff08;文件传输协议&#xff09;和SFTP&#xff08;安全文件传输协议&#xff09;成…
阅读更多...
前端图片在切换暗黑模式时太亮该怎么办？

前端图片在切换暗黑模式时太亮该怎么办？

通过css中的filter属性来实现&#xff0c;进行图片的色系反转、亮度、对比度调整等 1、invert 反转输入图像&#xff0c;值为 100% 则图像完全反转&#xff0c;值为 0% 则图像无变化 filter: invert(1); 2、blur 给元素应用高斯模糊效果。 filter: blur(5px); 3、brightnes…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023