大数据——Flink Broadcast State 广播状态

借酒温书

1、BroadcastState 的介绍 

播送形态(Broadcast State)是 Operator State 的一种特别范例。假如我们需求将设置 、划定规矩等低吞吐变乱流播送到下流一切 Task 时，就能够利用 BroadcastState。下流的 Task 吸取那些设置、划定规矩并保留为 BroadcastState，一切Task 中的形态连结分歧，感化于另外一个数据流的策画中。
简朴了解：一个低吞吐量流包含一组划定规矩，我们念对去自另外一个流的一切元素基于此划定规矩停止评价。
场景：静态更新策画划定规矩。
播送形态取其他操纵符形态的区分正在于:

• 它有一个 map 格局，用于界说存储构造
  
• 它仅对具有播送流战非播送流输进的特定操纵符可用
  
• 如许的操纵符能够具有不同称号的多个播送形态
  

 
2、BroadcastState 操纵流程

3、案例完成

• 从端心读与Json数据做为变乱流
  
• 从Mysql读与数据做为播送流
  
• 联系关系播送流战变乱流
  
• 婚配对应的用户疑息
  

1. package cn.kgc.broadcast
3. import java.sql.{Connection, DriverManager, PreparedStatement}
5. import com.alibaba.fastjson.JSON
6. import org.apache.flink.api.common.state.{BroadcastState, MapStateDescriptor}
7. import org.apache.flink.configuration.Configuration
8. import org.apache.flink.streaming.api.datastream.BroadcastStream
9. import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
10. import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
11. import org.apache.flink.streaming.api.scala._
12. import org.apache.flink.util.Collector
14. // (001,'tom',18,'北京',15830010002)
15. // 界说样例类 承受 MySQL的用户数据
16. case class BaseUserInfo(id:Long,name:String,age:Int,city:String,phone:Long)
18. // user_id、user_name、user_addrss、behaviour、url
19. // 输出数据范例
20. case class UserVisitInfo(id:Long,name:String,city:String,behaviour:String,url:String)
22. // 完成播送ProcessFunction
23. class MyBroadcastFunc extends BroadcastProcessFunction[String,(Long, BaseUserInfo),UserVisitInfo]{
25.   lazy val mapStateDes = new MapStateDescriptor[Long, BaseUserInfo]("mapState",classOf[Long],classOf[BaseUserInfo])
27.   // 处置的是日记流中的每条数据
28.   override def processElement(value: String, ctx: BroadcastProcessFunction[String, (Long, BaseUserInfo), UserVisitInfo]#ReadOnlyContext, out: Collector[UserVisitInfo]): Unit = {
29.     // {"user_id":"001","ts":"2021-07-10 11:10:05","behaviour":"browse","url":"https://www.tb1.com/1.html"}
30.     val user_id = JSON.parseObject(value).getLong("user_id")
31.     val behaviour = JSON.parseObject(value).getString("behaviour")
32.     val url = JSON.parseObject(value).getString("url")
34.     val mapState = ctx.getBroadcastState(mapStateDes)
35.     val userInfo = mapState.get(user_id)
37.     out.collect(UserVisitInfo(user_id,userInfo.name,userInfo.city,behaviour,url))
39.   }
41.   // 处置的是播送流的每一个值
42.   override def processBroadcastElement(value: (Long, BaseUserInfo), ctx: BroadcastProcessFunction[String, (Long, BaseUserInfo), UserVisitInfo]#Context, out: Collector[UserVisitInfo]): Unit = {
43.     val mapState: BroadcastState[Long, BaseUserInfo] = ctx.getBroadcastState(mapStateDes)
44.     mapState.put(value._1,value._2)
45.   }
46. }
49. class UserSourceFunc extends RichParallelSourceFunction[BaseUserInfo]{
51.   var conn:Connection = _
52.   var statement: PreparedStatement = _
53.   var flag:Boolean = true
55.   override def open(parameters: Configuration): Unit = {
56.     conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?characterEncoding=utf-8&serverTimezone=UTC","root","liu911223")
57.     statement = conn.prepareStatement("select * from base_user")
58.   }
60.   override def run(ctx: SourceFunction.SourceContext[BaseUserInfo]): Unit = {
61.     while (flag){
62.       Thread.sleep(5000)
63.       val resultSet = statement.executeQuery()
64.       while (resultSet.next()){
65.         val id = resultSet.getLong(1)
66.         val name = resultSet.getString(2)
67.         val age = resultSet.getInt(3)
68.         val city = resultSet.getString(4)
69.         val phone = resultSet.getLong(5)
70.         ctx.collect(BaseUserInfo(id,name,age,city,phone))
71.       }
72.     }
73.   }
75.   override def cancel(): Unit = {
76.     flag = false
77.   }
79.   override def close(): Unit = {
80.     if (statement != null) statement.close()
81.     if (conn != null) conn.close()
82.   }
83. }
84. object BroadcastDemo01 {
85.   def main(args: Array[String]): Unit = {
86.     val env = StreamExecutionEnvironment.getExecutionEnvironment
87.     env.setParallelism(1)
89.     // 界说为KV，一圆里是为了播送的时分界说为map，另外一圆里是为了做联系关系操纵
90.     val userBaseDS: DataStream[(Long, BaseUserInfo)] = env.addSource(new UserSourceFunc)
91.       .map(user => (user.id, user))
92.     val mapStateDes = new MapStateDescriptor[Long, BaseUserInfo]("mapState",classOf[Long],classOf[BaseUserInfo])
93.     val broadCastStream: BroadcastStream[(Long, BaseUserInfo)] = userBaseDS.broadcast(mapStateDes)
95.     // 日记JSON数据
96.     val dataInfoDS: DataStream[String] = env.socketTextStream("master",1314)
98.     dataInfoDS.connect(broadCastStream)
99.       .process(new MyBroadcastFunc)
100.       .print()
102.     env.execute()
103.   }
104. }

复造代码

免责声明：假如进犯了您的权益，请联络站少，我们会实时删除侵权内乱容，感谢协作！