flink重温笔记(十):Flink 高级 API 开发——flink 四大基石之 State(涉及Checkpoint)

news/2024/4/24 13:19:26/文章来源:https://blog.csdn.net/m0_60732994/article/details/136480132

Flink学习笔记

前言:今天是学习 flink 的第 10 天啦!学习了 flink 四大基石之 State (状态),主要是解决大数据领域增量计算的效果,能够保存已经计算过的结果数据状态!重点学习了 state 的类型划分和应用,以及 TTL 原理和应用,即数据状态也会过期和定期清除的问题,以及广播流数据的企业应用场景,结合自己实验猜想和代码实践,总结了很多自己的理解和想法,希望和大家多多交流!

Tips:广州回南天色佳,学习 state 意更浓。心随知识飘然去,智慧之舟破浪中。越来越有状态,明天也要继续努力!


文章目录

  • Flink学习笔记
    • 三、Flink 高级 API 开发
      • 3. State
        • 3.1 State 应用场景
        • 3.2 State 类型划分
          • 3.2.1 Keyed State 键控状态
            • (1) 特点
            • (2) 保存的数据结构
            • (3) 案例演示
          • 3.2.1 Operate State 算子状态
            • (1) 特点
            • (2) 保存的数据结构
            • (3) 案例演示
        • 3.3 State TTL 状态有效期
          • 3.3.1 功能用法
            • (1) TTL 的更新策略
            • (2) 状态数据过期但未被清除时
            • (3) 过期数据的清除
          • 3.3.2 案例演示
        • 3.4 Broadcast State
          • 3.4.1 应用场景
          • 3.4.2 注意事项
          • 3.4.3 案例演示
          • 3.4.4 BroadcastState 执行思路梳理

三、Flink 高级 API 开发

3. State

简介:State(状态)是基于 Checkpoint(检查点)来完成状态持久化,在 Checkpoint 之前,State 是在内存中(变量),在 Checkpoint 之后,State 被序列化永久保存,支持存储方式:File,HDFS,S3等。

3.1 State 应用场景
  • (1)去重
  • (2)窗口计算
  • (3)机器学习/深度学习
  • (4)访问历史数据

3.2 State 类型划分
  • 基本类型划分
    • Keyed State(键控状态)
    • Operate State(算子状态)
  • 存在方式划分
    • raw State (原始状态):原始状态
    • managed State(托管状态):Flink 自动管理的 State,实际生产推荐使用
  • 原始状态和托管状态区别:
状态管理方式数据结构使用场景
Managed StateFlink Runtime 管理自动存储,自动恢复内存管理可自动优化Value,List,Map…大多数情况下均可使用
Raw State需要用户自己管理,需要自己序列化字节数组自定义Operator时使用
3.2.1 Keyed State 键控状态
(1) 特点
  • 1- 只能用于 keyby 后的数据流
  • 2- 一个 key 只能属于 一个key State
(2) 保存的数据结构
  • Keyed State 通过 RuntimeContext 访问,这需要 Operator 是一个RichFunction。
  • 1- ValueState:单值状态
  • 2- ListState:列表状态
  • 3- ReducingState:传入reduceFunction,单一状态值
  • 4- MapState<UK, UV>:状态值为 map
(3) 案例演示

例子:词频统计,不要用 sum,而是用 reduce,然后 ValueState

package cn.itcast.day10.state;import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** @author lql* @time 2024-03-03 16:01:44* @description TODO:演示 keyedState 的使用*/
public class KeyedStateDemo {public static void main(String[] args) throws Exception {//todo 1)创建flink流处理的运行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//todo 2)开启checkpointenv.enableCheckpointing(5000);//todo 3)构建数据源DataStreamSource<String> lines = env.socketTextStream("node1", 9999);//todo 4)单词拆分SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndNum = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String line) throws Exception {String[] data = line.split(",");return Tuple2.of(data[0], Integer.parseInt(data[1]));}});//todo 5)分流操作KeyedStream<Tuple2<String, Integer>, String> keyedDataStream = wordAndNum.keyBy(t -> t.f0);//todo 6) 聚合操作(自定义state方式实现)SingleOutputStreamOperator<Tuple2<String, Integer>> reduceState = keyedDataStream.reduce(new RichReduceFunction<Tuple2<String, Integer>>() {// todo 6.1 定义 state 对象private ValueState<Tuple2<String, Integer>> valueState = null;// 初始化资源@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);// todo 6.2 实例化 state 对象valueState = getRuntimeContext().getState(new ValueStateDescriptor<Tuple2<String, Integer>>("reduceState",TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {})));}@Overridepublic Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {// state 存储的历史数据// todo 6.3 获取state 对象Tuple2<String, Integer> result = valueState.value();if (result == null) {result = Tuple2.of(value1.f0, value1.f1);}Tuple2<String, Integer> resultSum = Tuple2.of(result.f0, result.f1 + value2.f1);// 之前没有重写状态:Tuple2.of(value1.f0, value1.f1 + value2.f1);// 因为我们这里的 value1 已经从 state 中获取到 result了,所以重写状态的方法是,就用result!// todo 6.4 更新 state 对象valueState.update(resultSum);return resultSum;}// 释放资源@Overridepublic void close() throws Exception {super.close();// 获取 state 的数据Tuple2<String, Integer> value = valueState.value();System.out.println("=======释放资源的时候打印 state 数据========");System.out.println(value);}});//todo 6)打印测试reduceState.print();//todo 7)运行env.execute();}
}

结果:

输入:
hadoop,1
hadoop,2输出:
8> (hadoop,1)
8> (hadoop,3)

总结:

  • 1- state 类似于一个数据库,可以从中获取之前计算过的数据
  • 2- 定义 state 对象,初始值为 null
  • 3- 实例化 state 对象,getRuntimeContext().getState(new ValueStateDescriptor())
  • 4- 获取 state 中的数据:状态.value()
  • 5- 更新 state 中的数据:状态.update(数据流)

3.2.1 Operate State 算子状态
(1) 特点
  • 1- 一个算子状态仅与一个算子实例绑定
  • 2- 算子状态可以用于所有算子,但常见是 Source
(2) 保存的数据结构
  • Operator State 需要自己实现 CheckpointedFunction 或 ListCheckpointed 接口
  • 1- ListState
  • 2- BroadcastState<K,V>
(3) 案例演示

例子:使用 OperatorState 进行演示基于类似于 kafka 消费数据的功能

package cn.itcast.day10.state;/*** @author lql* @time 2024-03-03 17:21:50* @description TODO:使用OperatorState进行演示基于类似于kafka消费数据的功能*/import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;import java.util.List;
import java.util.concurrent.TimeUnit;/*** 实现步骤:* 1)初始化flink流式程序的运行环境* 2)设置并行度为1* 3)启动checkpoint* 4)接入数据源* 5)打印测试* 6)启动作业*/
public class OperatorStateDemo {public static void main(String[] args) throws Exception {// todo 1) 初始化环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// todo 2) 设置并行度为 1env.setParallelism(1);// todo 3) 启动checkpoint机制env.enableCheckpointing(4000);// todo 4) 接入数据源DataStreamSource<Integer> source  = env.addSource(new MySourceWithState());SingleOutputStreamOperator<String> result = source.map(new MapFunction<Integer, String>() {@Overridepublic String map(Integer integer) throws Exception {return integer.toString();}});//TODO 5)打印测试result.printToErr();//TODO 6)启动作业env.execute();}private static class MySourceWithState extends RichSourceFunction<Integer> implements CheckpointedFunction {//定义成员变量是否循环生成数据private Boolean isRunning = true;private Integer currentCounter = 0;//定义ListState保存结果数据。保存offset的累加值private ListState<Integer> listState = null;/*** 将state中的数据持久化存储到文件中(每4秒钟进行一次快照,将state数据存储到hdfs)* @param functionSnapshotContext* @throws Exception*/@Overridepublic void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {System.out.println("调用snapshotState方法。。。。。。。。");// 清除历史状态存储的历史数据,this引用对象的成员变量this.listState.clear();//将最新的累加值添加到状态中this.listState.add(this.currentCounter);}/*** 初始化 state 对象* @param context* @throws Exception*/@Overridepublic void initializeState(FunctionInitializationContext context) throws Exception {System.out.println("调用initializeState方法。。。。。。。。");//初始化一个listStateOperatorStateStore stateStore  = context.getOperatorStateStore();listState = stateStore.getListState(new ListStateDescriptor<>("operator-states",TypeInformation.of(new TypeHint<Integer>() {})));// 获取历史数据for (Integer counter  : this.listState.get()) {//将历史存储的累加值取出来赋值给当前累加值变量this.currentCounter = counter;}//清除状态中存储的历史数据this.listState.clear();}/*** 生产数据* @param sourceContext* @throws Exception*/@Overridepublic void run(SourceContext<Integer> sourceContext) throws Exception {while (isRunning){currentCounter ++;sourceContext.collect(currentCounter);TimeUnit.SECONDS.sleep(1);if (this.currentCounter == 10){System.out.println("手动抛出异常"+(1/0));}}}/*** 取消生产数据*/@Overridepublic void cancel() {isRunning = false;}}
}

结果:

调用initializeState方法。。。。。。。。
1
2
3
4
调用snapshotState方法。。。。。。。。
5
6
7
8
调用snapshotState方法。。。。。。。。
9
10
调用initializeState方法。。。。。。。。
9
调用snapshotState方法。。。。。。。。
10<==10这里出现异常了,所以保存state中的数据是9,初始化方法后恢复 9+1 数据,然后 10 也保存到state中了==>调用initializeState方法。。。。。。。。
10
调用snapshotState方法。。。。。。。。
调用initializeState方法。。。。。。。。
11
12
13
14
调用snapshotState方法。。。。。。。。
15
16
17

总结:

  • 1- 定义数据源要继承 RichSourceFunction 父类,实现 CheckpointedFunction 接口
  • 2- 初始化 listState:getOperatorStateStore.getListState(new ListStateDescriptor<>)
  • 3- 获取历史数据:因为是listState,所以用循环 get()
  • 4- 历史值赋予当前值,清空历史数据 clear()
  • 5- 检查点之后,恢复数据就是 [ 历史数据 + 1 ]

3.3 State TTL 状态有效期

举例子:更新策略着眼于是更新日期是在哪个时候,

​ 而这里设置停留时间 .newBuilder(Time.seconds(1)) 是指状态保存多长时间,

​ 时间一过状态数据就标记过期(设置时间要比 checkpoint 时间长,才能保证 checkpoint 顺利持久化),

清除策略着眼于过期数据清理是在哪个时候

应用场景:使用 flink 进行实时计算中,会遇到一些状态数不断累积,导致状态量越来越大的情形


3.3.1 功能用法
(1) TTL 的更新策略
  • 只在创建和写入时更新:.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)

  • 读取和写入时也会更新:.setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)

(2) 状态数据过期但未被清除时
  • 不返回过期数据:.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
  • 返回未被清除的过期数据:.setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp)
(3) 过期数据的清除
  • 关闭后台自动清理过期数据:disableCleanupInBackground()

  • 全量快照时清理:cleanupFullSnapshot()

    • 这种策略在 RocksDBStateBackend 的增量 checkpoint 模式下无效。
  • 增量快照时清理:.cleanupIncrementally(10, true)

    • 第一个是每次清理时检查状态的条目数,在每个状态访问时触发。

      第二个参数表示是否在处理每条记录时触发清理。

  • 后台自动清理(RocksDB state backend 存储):.cleanupInRocksdbCompactFilter(1000)

    • 默认后台清理策略会每处理 1000 条数据进行一次,

    • 表示在 RocksDB 的compaction过程中,每删除1000个键值对,就会执行一次TTL过期的键值对的清理。

    • RocksDB 的 compaction是一个过程,它会合并多个小的数据库文件(SSTables)成一个大的文件


3.3.2 案例演示

例子1:10s 读取一行数据,checkpoint 60s,state 时间 5 s

package cn.itcast.day10.state;import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.util.Collector;import java.io.BufferedReader;
import java.io.FileReader;
import java.util.concurrent.TimeUnit;/*** @author lql* @time 2024-03-04 18:53:52* @description TODO*/
public class StateWordCount {public static void main(String[] args) throws Exception {// todo 1) 使用工具类将传入的参数解析为对象final ParameterTool parameters = ParameterTool.fromArgs(args);// todo 2) 初始化环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);// todo 3) 将传入的参数解析成参数注册到作业中env.getConfig().setGlobalJobParameters(parameters);// todo 4) 开启 checkpoint,一致性语义env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);// todo 5) 接入数据源DataStreamSource<String> lines = env.addSource(new SourceFunctionFile());lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {@Overridepublic void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {String[] data = s.split(" ");for (String item : data) {collector.collect(new Tuple2<>(item, 1));}}}).keyBy(t -> t.f0).flatMap(new WordCountFlatMap()).printToErr();// todo 6)启动作业env.execute();}private static class SourceFunctionFile extends RichSourceFunction<String> {// 定义是否继续生成数据标记private Boolean isRunning = true;@Overridepublic void run(SourceContext<String> sourceContext) throws Exception {BufferedReader bufferedReader = new BufferedReader(new FileReader("D:\\IDEA_Project\\BigData_Java\\flinkbase_pro\\data\\input\\wordcount.txt"));while (isRunning) {String line = bufferedReader.readLine();if (StringUtils.isBlank(line)) {continue;}sourceContext.collect(line);TimeUnit.SECONDS.sleep(10);}}@Overridepublic void cancel() {isRunning = false;}}// 自定义聚合private static class WordCountFlatMap extends RichFlatMapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>> {// 定义状态private ValueState<Tuple2<String, Integer>> valueState = null;@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);//创建ValueStateDescriptorValueStateDescriptor valueStateDescriptor = new ValueStateDescriptor("StateWordCount",TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {}));//配置stateTTLStateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.seconds(5))  // 存活时间.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) // 创建或更新的时候修改时间.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) // 永不返回过期的数据.cleanupFullSnapshot() // 全量快照时进行清理.build();// 激活 stateTTLvalueStateDescriptor.enableTimeToLive(ttlConfig);// 实例化 valueState 对象valueState = getRuntimeContext().getState(valueStateDescriptor);}@Overridepublic void close() throws Exception {super.close();}@Overridepublic void flatMap(Tuple2<String, Integer> value, Collector<Tuple2<String, Integer>> collector) throws Exception {// 获取状态数据Tuple2<String, Integer> currentState = valueState.value();// 初始化 valueState 数据if (currentState == null) {currentState = new Tuple2<>(value.f0,0);}// 累加单词的次数Tuple2<String, Integer> newState = new Tuple2<>(currentState.f0, currentState.f1 + value.f1);// 更新valueStatevalueState.update(newState);// 返回累加后的结果collector.collect(newState);}}
}

数据源:

Total time BUILD SUCCESS
Final Memory Finished at
Total time BUILD SUCCESS
Final Memory Finished at
Total time BUILD SUCCESS
Final Memory Finished at
BUILD SUCCESS
BUILD SUCCESS
BUILD SUCCESS
BUILD SUCCESS
BUILD SUCCESS

结果:没有累加历史值

======110s=====
(Total,1)
(time,1)
(BUILD,1)
(SUCCESS,1)======210s=====
(Final,1)
(Memory,1)
(Finished,1)
(at,1)======310s=====
(Total,1)
(time,1)
(BUILD,1)
(SUCCESS,1)

例子2:10s 读取一行数据,checkpoint 5s,state 时间 20 s 或者 21s

结果:触发累加历史值

======110s=====
(Total,1)
(time,1)
(BUILD,1)
(SUCCESS,1)======210s=====
(Final,1)
(Memory,1)
(Finished,1)
(at,1)======310s=====
(Total,2)
(time,2)
(BUILD,2)
(SUCCESS,2)

例子3:10s 读取一行数据,checkpoint 5s,state 时间 19s

结果:没有累加历史值


  • 总结1: 从例子1-3,因为第1个Total距第2个Total 20s,故 state 设置要尽可能大20s,设置太小来不及遇见第二个就过期了。


例子4:10s 读取一行数据,checkpoint 10s,state 时间 20 s

结果:触发累加历史值


例子5:10s 读取一行数据,checkpoint 60s,state 时间 20s

结果:触发累加历史值


例子6:10s 读取一行数据,checkpoint 20s,state 时间 20 s

结果:没有累加历史值


  • 总结2:从例子 4-6,checkpoint 比 state 建议小很多或大很多,不然 [10-20] 有些不能触发,[20-60] 有些能触发。


例子7:10s 读取一行数据,checkpoint 19s,state 时间 20s(之前不能触发)

(此时设置.setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp) // 返回过期但未被清除的数据)

结果:触发累加历史值


例子8:10s 读取一行数据,checkpoint 60s,state 时间 5 s(之前state小于20s不能触发)

结果:触发累加历史值


例子9:10s 读取一行数据,checkpoint 60s,state 时间 20s(之前可以触发)

(此时设置.setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp) // 返回过期但未被清除的数据)

结果:触发累加历史值


  • 总结3:从例子7-9,设置 ReturnExpiredIfNotCleanedUp 之前可以触发的不可以触发的,都可以触发了!


总结:

  • 1- state 设置要尽可能大于相同数据间隔s,设置太小来不及遇见第二个就过期了。
  • 2- checkpoint 比 state 建议小很多或大很多
  • 3- 设置 ReturnExpiredIfNotCleanedUp 可以避免不能触发
  • 4- BufferedReader 去 new 一个 FileReader 可以读取文件
  • 5- 激活 stateTTL:valueStateDescriptor.enableTimeToLive(ttlConfig)

3.4 Broadcast State
3.4.1 应用场景
  • 1-动态更新计算规则: 如事件流需要根据最新规则进行计算,可将规则作为广播状态广播到下游Task。
  • 2-实时增加额外字段: 如事件流需要实时增加用户基础信息,可将基础信息作为广播状态到下游Task。

3.4.2 注意事项
  • Broadcast State 是 Map 类型,即 K-V 类型

  • Broadcast State 只有在广播的一侧,即在重写方法:processBroadcastElement 方法中可修改,另一个方法只读

  • Broadcast State 在 checkpoint 时,每个 Task 都会 checkpoint(持久化)广播状态。

  • Broadcast State 在运行时保存在内存中,(flink 1.13)还不能保存在 Rocked State Backend


3.4.3 案例演示

例子:公司有10个广告位, 其广告的内容(描述和图片)会经常变动(广告到期,更换广告等)

package cn.itcast.day10.state;/*** @author lql* @time 2024-03-05 13:54:54* @description TODO*/import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;import java.util.*;
import java.util.concurrent.TimeUnit;/*** 广播状态流演示* 需求:公司有10个广告位, 其广告的内容(描述和图片)会经常变动(广告到期,更换广告等)* 实现:* 1)通过socket输入广告id(事件流)* 2)关联出来广告的信息打印出来,就是广告发生改变的时候,能够感知到(规则流)*/
public class BroadcastStateDemo {public static void main(String[] args) throws Exception {//todo 1)初始化flink流处理的运行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//todo 2) 设置checkpoint周期运转env.enableCheckpointing(5000L);//todo 3) 构建数据流 (我输入的是字符串数字,需要转化为整数类型)DataStreamSource<String> lines = env.socketTextStream("node1", 9999);SingleOutputStreamOperator<Integer> adIdDataStream = lines.map(new MapFunction<String, Integer>() {@Overridepublic Integer map(String value) throws Exception {return Integer.parseInt(value);}});// todo 4) 构建规则流(广告流)DataStreamSource<Map<Integer, Tuple2<String, String>>> adSourceStream = env.addSource(new MySourceForBroadcastFunction());adSourceStream.print("最新的广告信息>>>");// todo 5) 将规则流(广告流)转化为广播流==> 这里和之前的广播是不一样的!// todo 5.1 定义规则流(广告流)的描述器MapStateDescriptor<Integer, Tuple2<String, String>> mapStateDescriptor = new MapStateDescriptor<Integer, Tuple2<String, String>>("broadcaststate",TypeInformation.of(new TypeHint<Integer>() {}),TypeInformation.of(new TypeHint<Tuple2<String,String>>(){}));// todo 5.2 运用描述器将(广告流)转化为广播流BroadcastStream<Map<Integer, Tuple2<String, String>>> broadcastStream = adSourceStream.broadcast(mapStateDescriptor);// todo 6) 将数据流和规则流合并在一起BroadcastConnectedStream<Integer, Map<Integer, Tuple2<String, String>>> connectedStream = adIdDataStream.connect(broadcastStream);// todo 7) 对关联后的数据做拉宽操作SingleOutputStreamOperator<Tuple2<String, String>> result = connectedStream.process(new MyBroadcastProcessFunction());// todo 8) 打印结果数据result.printToErr("拉宽后的结果>>>");//todo 9)启动作业env.execute();}/*** 自定义规则数据,注意 返回类型是 Map<K,V>*/private static class MySourceForBroadcastFunction implements SourceFunction<Map<Integer, Tuple2<String,String>>> {private final Random random = new Random();private final List<Tuple2<String, String>> ads = Arrays.asList(Tuple2.of("baidu", "搜索引擎"),Tuple2.of("google", "科技大牛"),Tuple2.of("aws", "全球领先的云平台"),Tuple2.of("aliyun", "全球领先的云平台"),Tuple2.of("腾讯", "氪金使我变强"),Tuple2.of("阿里巴巴", "电商龙头"),Tuple2.of("字节跳动", "靠算法出名"),Tuple2.of("美团", "黄色小公司"),Tuple2.of("饿了么", "蓝色小公司"),Tuple2.of("瑞幸咖啡", "就是好喝"));private boolean isRun = true;@Overridepublic void run(SourceContext<Map<Integer, Tuple2<String, String>>> sourceContext) throws Exception {while (isRun){// 定义一个 HashMap,用来存储键值对HashMap<Integer, Tuple2<String,String>> map = new HashMap<>();int keyCounter = 0;for (int i = 0; i < ads.size(); i++) {keyCounter++;map.put(keyCounter,ads.get(random.nextInt(ads.size())));}sourceContext.collect(map);TimeUnit.SECONDS.sleep(5L);}}@Overridepublic void cancel() {isRun = false;}}private static class MyBroadcastProcessFunction extends BroadcastProcessFunction<Integer,Map<Integer, Tuple2<String, String>>, Tuple2<String, String>> {//定义state的描述器MapStateDescriptor<Integer, Tuple2<String, String>> mapStateDescriptor = new MapStateDescriptor<Integer, Tuple2<String, String>>("broadcaststate",TypeInformation.of(new TypeHint<Integer>() {}),TypeInformation.of(new TypeHint<Tuple2<String, String>>() {}));/*** 这个方法只读,用来拉宽操作* @param integer* @param readOnlyContext* @param collector* @throws Exception*/@Overridepublic void processElement(Integer integer, ReadOnlyContext readOnlyContext, Collector<Tuple2<String, String>> collector) throws Exception {//只读操作,意味着只能读取数据,不能修改数据,根据广告id获取广告信息ReadOnlyBroadcastState<Integer, Tuple2<String, String>> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);//根据广告id获取广告信息Tuple2<String, String> tuple2 = broadcastState.get(integer);//判断广告信息是否关联成功if(tuple2 != null) {collector.collect(tuple2);}}/*** 可写的,用来更新state的数据* @param integerTuple2Map* @param context* @param collector* @throws Exception*/@Overridepublic void processBroadcastElement(Map<Integer, Tuple2<String, String>> integerTuple2Map, Context context, Collector<Tuple2<String, String>> collector) throws Exception {//先读取state数据BroadcastState<Integer, Tuple2<String, String>> broadcastState = context.getBroadcastState(mapStateDescriptor);// 删除历史状态数据broadcastState.clear();//将最新获取到的广告信息进行广播操作broadcastState.putAll(integerTuple2Map);}}
}

结果:

最新的广告信息>>>:7> {1=(aws,全球领先的云平台), 2=(aliyun,全球领先的云平台), 3=(阿里巴巴,电商龙头), 4=(aws,全球领先的云平台), 5=(瑞幸咖啡,就是好喝), 6=(瑞幸咖啡,就是好喝), 7=(美团,黄色小公司), 8=(aws,全球领先的云平台), 9=(腾讯,氪金使我变强), 10=(字节跳动,靠算法出名)}最新的广告信息>>>:8> {1=(美团,黄色小公司), 2=(饿了么,蓝色小公司), 3=(aws,全球领先的云平台), 4=(baidu,搜索引擎), 5=(aws,全球领先的云平台), 6=(baidu,搜索引擎), 7=(美团,黄色小公司), 8=(字节跳动,靠算法出名), 9=(瑞幸咖啡,就是好喝), 10=(腾讯,氪金使我变强)}======我在终端输入了 3,刚好对应上面一条信息的 3 位置========
拉宽后的结果>>>:4> (aws,全球领先的云平台)最新的广告信息>>>:1> {1=(饿了么,蓝色小公司), 2=(aliyun,全球领先的云平台), 3=(google,科技大牛), 4=(瑞幸咖啡,就是好喝), 5=(美团,黄色小公司), 6=(baidu,搜索引擎), 7=(google,科技大牛), 8=(google,科技大牛), 9=(google,科技大牛), 10=(腾讯,氪金使我变强)}最新的广告信息>>>:2> {1=(google,科技大牛), 2=(饿了么,蓝色小公司), 3=(字节跳动,靠算法出名), 4=(aliyun,全球领先的云平台), 5=(aws,全球领先的云平台), 6=(aws,全球领先的云平台), 7=(google,科技大牛), 8=(google,科技大牛), 9=(美团,黄色小公司), 10=(aliyun,全球领先的云平台)}======我在终端输入了 1,刚好对应上面一条信息的 1 位置========
拉宽后的结果>>>:5> (google,科技大牛)

总结:

  • 1- 这里的规则流转化为广播流操作,和之前广播分区不一样
    • 数据广播到各个分区:数据.Brocast()
    • 数据流转化为广播流:数据.Brocast(描述器)
  • 2- BroadcastProcessFunction 方法中重写的 processElement 只读方法,用来拉宽操作
    • 获取数据+关联数据(判非空后收集)
  • 3- BroadcastProcessFunction 方法中重写的 processBroadcastElement 可修改方法,
    • 读取 state 数据:getBroadcastState()
    • 删除历史数据:clear()
    • 将新的实例添加到广播状态中:putAll()

3.4.4 BroadcastState 执行思路梳理

在这里插入图片描述


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

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

相关文章

力扣hot100:240.搜索二维矩阵II(脑子)

吉大21级算法分析与设计的一道大题&#xff0c;由于每一行都是排好序的直接逐行二分 可以达到&#xff1a;O(mlogn)。但是这里追求更广的思路可以使用其他方法。 矩阵四分&#xff1a; 在矩阵中用中心点比较&#xff0c;如果target大于中心点的值&#xff0c;则由于升序排列&am…

超全Chat GPT论文修改指令

文献综述指令润色修改指令论文选题指令论文大指令研究理论指令论文致谢指令参考文献指令论文润色整体逻辑论文整体优化提问指令 1&#xff0e;文献综述指令 请你帮我写一份关于&#xff08;研究主题&#xff09;的文献综述。我的论文选题方向是 XXXX &#xff0c;我已经找到了…

一次电脑感染Synaptics Pointing Device Driver病毒的经历,分享下经验

没想到作为使用电脑多年的老司机也会电脑中病毒&#xff0c;周末玩电脑的时候突然电脑很卡&#xff0c;然后自动重启&#xff0c;奇怪&#xff0c;之前没出现这个情况。 重启后电脑开机等了几十秒&#xff0c;打开任务管理器查看开机进程&#xff0c;果然发现有个Synaptics Po…

C语言实现回调函数

C语言实现回调函数 一、回调函数概念1.1 什么叫函数指针 二、回调函数案例 一、回调函数概念 回调函数就是一个被作为参数传递的函数。在C语言中&#xff0c;回调函数只能使用函数指针实现&#xff0c;在C、Python、ECMAScript等更现代的编程语言中还可以使用仿函数或匿名函数…

应用层DDoS防护:理解、必要性与实现策略

一、应用层简介 应用层&#xff0c;也称作第七层&#xff0c;是OSI&#xff08;开放系统互联&#xff09;模型中的最高层。在这一层&#xff0c;数据以特定的应用程序协议格式进行传输&#xff0c;如HTTP、FTP、SMTP等。应用层的主要职责是为用户提供网络服务&#xff0c;如文…

[嵌入式系统-37]:龙芯1B 开发学习套件 -6-协处理器CP0之CPU异常处理与外部中断控制器的中断处理

目录 一、MPIS CPU Core与32个异常exception 1.1 龙芯1B的MIPS CPU IP Core 1.2 MIP32指令系统 1.3 MIPS CPU寄存器 1.4 MIPS CPU的异常向量与异常向量号 1.5 龙芯异常exception与中断interrupt的区别 二、协议处理器CP0的中断控制与8个中断 2.1 CP0概述 2.2 协处理器…

【C++精简版回顾】18.文件操作

1.文件操作头文件 2.操作文件所用到的函数 1.文件io 1.头文件 #include<fstream> 2.打开文件 &#xff08;1&#xff09;函数名 文件对象.open &#xff08;2&#xff09;函数参数 /* ios::out 可读 ios::in 可…

类与对象(三)--static成员、友元

文章目录 1.static成员1.1概念&#x1f3a7;面试题✒️1.2static的特性&#x1f3a7;1.3思考&#x1f3a7; 2.友元2.1什么是友元&#xff1f;&#x1f3a7;2.2两种友元关系&#xff1a;&#x1f3a7; 1.static成员 1.1概念&#x1f3a7; &#x1f50e; static关键字用于声明类…

IO接口 2月5日学习笔记

1.fgetc 用于从文件中读取一个字符&#xff0c;fgetc 函数每次调用将会返回当前文件指针所指向的字符&#xff0c;并将文件指针指向下一个字符。 int fgetc(FILE *stream); 功能: 从流中读取下一个字符 参数: stream:文件流指针 返回值: …

【Docker】若依ruoyi项目部署

一 搭建局域网 1 # 搭建net-ry局域网&#xff0c;用于部署若依项目docker network create net-ry --subnet172.68.0.0/16 --gateway172.68.0.1 # 注意1&#xff1a;关闭宿主机的防火墙&#xff0c;否者容器内部的MySQL、redis等服务&#xff0c;外部访问不了&#xff1b;开放…

Linux 文件系列:深入理解文件描述符fd,重定向,自定义shell当中重定向的模拟实现

Linux 文件系列:深入理解文件fd,重定向,自定义shell当中重定向的模拟实现 一.预备知识二.回顾C语言中常见的文件接口跟重定向建立联系1.fopen函数的介绍2.fclose函数的介绍3.代码演示1.以"w"(写)的方式打开2.跟输出重定向的联系3.以 "a"(追加)的方式打开4.…

导出谷歌gemma模型为ONNX

参考代码如下&#xff08;从GitHub - luchangli03/export_llama_to_onnx: export llama to onnx修改而来&#xff0c;后面会合入进去&#xff09; 模型权重链接参考&#xff1a; https://huggingface.co/google/gemma-2b-it 可以对modeling_gemma.py进行一些修改(transforme…

革命文物的新征程:SpringBoot实践

✍✍计算机编程指导师 ⭐⭐个人介绍&#xff1a;自己非常喜欢研究技术问题&#xff01;专业做Java、Python、微信小程序、安卓、大数据、爬虫、Golang、大屏等实战项目。 ⛽⛽实战项目&#xff1a;有源码或者技术上的问题欢迎在评论区一起讨论交流&#xff01; ⚡⚡ Java实战 |…

【QT】定时器事件应用

public&#xff1a;//定时器void timerEvent(QTimerEvent *);private:int id;#include <QTimerEvent> //QWidget 默认不追踪鼠标事件 mylabel::mylabel(QWidget *parent) : QLabel(parent) {this->setMouseTracking(true);//启动定时器//参数1&#xff1a;触发定时器的…

electron+vue3全家桶+vite项目搭建【29】封装窗口工具类【3】控制窗口定向移动

文章目录 引入实现效果思路声明通用的定位对象主进程模块渲染进程测试效果 引入 demo项目地址 窗口工具类系列文章&#xff1a; 封装窗口工具类【1】雏形 封装窗口工具类【2】窗口组&#xff0c;维护窗口关系 封装窗口工具类【3】控制窗口定向移动 很多时候&#xff0c;我们想…

2024 AI 辅助研发的新纪年

随着人工智能技术的持续发展与突破&#xff0c;2024年AI辅助研发正成为科技界和工业界瞩目的焦点。从医药研发到汽车设计&#xff0c;从软件开发到材料科学&#xff0c;AI正逐渐渗透到研发的各个环节&#xff0c;变革着传统的研发模式。在这一背景下&#xff0c;AI辅助研发不仅…

【粉丝福利】探秘内部审计数字化之道:精准解析转型方法与成功路径

&#x1f33c;前言 内部审计是一种独立的、客观的确认和咨询活动&#xff0c;包括鉴证、识别和分析问题以及提供管理建议和解决方案。狭义的数字化转型是指将企业经营管理和业务操作的各种行为、状态和结果用数字的形式来记录和存储&#xff0c;据此再对数据进行挖掘、分析和应…

万字详解,Java实现低配版线程池

文章目录 1.什么是线程池2.线程池的优势3.原理4.代码编写4.1 阻塞队列4.2 ThreadPool线程池4.3 Worker工作线程4.4 代码测试 5. 拒绝策略5.1 抽象Reject接口5.2 BlockingQueue新增tryPut方法5.3 修改ThreadPool的execute方法5.4 ThreadPool线程池构造函数修改5.5 拒绝策略实现1…

运输层端口号、复用和分用的概念

运行在计算机上的进程用进程标识符PID来标志&#xff0c;但不同操作系统使用的进程标识符又不一样。 所以为了能让不同计算机上的进程进行网络通信&#xff0c;就需要统一的方式来进行标识&#xff0c;这就是TCP/IP体系的端口号&#xff0c;范围是0~65535&#xff0c;2字节 运…

详解float函数类型转换

函数描述 float([x]) 函数将数字或数字的字符串表示形式转换为与它等效的有符号浮点数。如果参数x是一个字符串&#xff08;十进制表示的数字串&#xff09;&#xff0c;数字前面可以添加符号来表示正数&#xff0c;或负数。符号和数字之间不能出现空格&#xff0c;但是符号前…