亲宝软件园·资讯

展开

[源码分析] 从实例和源码入手看 Flink 之广播 Broadcast

罗西的思考 人气:0
# [源码分析] 从实例和源码入手看 Flink 之广播 Broadcast ## 0x00 摘要 本文将通过源码分析和实例讲解,带领大家熟悉Flink的广播变量机制。 ## 0x01 业务需求 ### 1. 场景需求 对黑名单中的IP进行检测过滤。IP黑名单的内容会随时增减,因此是可以随时动态配置的。 该黑名单假设存在mysql中,Flink作业启动时候会把这个黑名单从mysql载入,作为一个变量由Flink算子使用。 ### 2. 问题 我们不想重启作业以便重新获取这个变量。所以就需要一个能够动态修改算子里变量的方法。 ### 3. 解决方案 使用广播的方式去解决。去做配置的动态更新。 广播和普通的流数据不同的是:广播流的1条流数据能够被算子的所有分区所处理,而数据流的1条流数据只能够被算子的某一分区处理。因此广播流的特点也决定适合做配置的动态更新。 ## 0x02 概述 广播这部分有三个难点:使用步骤;如何自定义函数;如何存取状态。下面就先为大家概述下。 ### 1. broadcast的使用步骤 - 建立MapStateDescriptor - 通过DataStream.broadcast方法返回广播数据流BroadcastStream - 通过DataStream.connect方法,把业务数据流和BroadcastStream进行连接,返回BroadcastConnectedStream - 通过BroadcastConnectedStream.process方法分别进行processElement及processBroadcastElement处理 ### 2. 用户自定义处理函数 - BroadcastConnectedStream.process接收两种类型的function:KeyedBroadcastProcessFunction 和 BroadcastProcessFunction - 两种类型的function都定义了processElement、processBroadcastElement抽象方法,只是KeyedBroadcastProcessFunction多定义了一个onTimer方法,默认是空操作,允许子类重写 - processElement处理业务数据流 - processBroadcastElement处理广播数据流 ### 3. Broadcast State - Broadcast State始终表示为MapState,即map format。这是Flink提供的最通用的状态原语。是托管状态的一种,托管状态是由Flink框架管理的状态,如ValueState, ListState, MapState等。 - 用户必须创建一个 `MapStateDescriptor`,才能得到对应的状态句柄。 这保存了状态名称, 状态所持有值的类型,并且可能包含用户指定的函数 - checkpoint的时候也会checkpoint broadcast state - Broadcast State只在内存有,没有RocksDB state backend - Flink 会将state广播到每个task,注意该state并不会跨task传播,对其修改仅仅是作用在其所在的task - downstream tasks接收到broadcast event的顺序可能不一样,所以依赖其到达顺序来处理element的时候要小心 ## 0x03. 示例代码 ### 1. 示例代码 我们直接从Flink源码入手可以找到理想的示例。 以下代码直接摘录 Flink 源码 StatefulJobWBroadcastStateMigrationITCase,我会在里面加上注释说明。 ```scala @Test def testRestoreSavepointWithBroadcast(): Unit = { val env = StreamExecutionEnvironment.getExecutionEnvironment env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) // 以下两个变量是为了确定广播流发出的数据类型,广播流可以同时发出多种类型的数据 lazy val firstBroadcastStateDesc = new MapStateDescriptor[Long, Long]( "broadcast-state-1", BasicTypeInfo.LONG_TYPE_INFO.asInstanceOf[TypeInformation[Long]], BasicTypeInfo.LONG_TYPE_INFO.asInstanceOf[TypeInformation[Long]]) lazy val secondBroadcastStateDesc = new MapStateDescriptor[String, String]( "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO) env.setStateBackend(new MemoryStateBackend) env.enableCheckpointing(500) env.setParallelism(4) env.setMaxParallelism(4) // 数据流,这里数据流和广播流的Source都是同一种CheckpointedSource。数据流这里做了一系列算子操作,比如flatMap val stream = env .addSource( new CheckpointedSource(4)).setMaxParallelism(1).uid("checkpointedSource") .keyBy( new KeySelector[(Long, Long), Long] { override def getKey(value: (Long, Long)): Long = value._1 } ) .flatMap(new StatefulFlatMapper) .keyBy( new KeySelector[(Long, Long), Long] { override def getKey(value: (Long, Long)): Long = value._1 } ) // 广播流 val broadcastStream = env .addSource( new CheckpointedSource(4)).setMaxParallelism(1).uid("checkpointedBroadcastSource") .broadcast(firstBroadcastStateDesc, secondBroadcastStateDesc) // 把数据流和广播流结合起来 stream .connect(broadcastStream) .process(new VerifyingBroadcastProcessFunction(expectedFirstState, expectedSecondState)) .addSink(new AccumulatorCountingSink) } } // 用户自定义的处理函数 class TestBroadcastProcessFunction extends KeyedBroadcastProcessFunction [Long, (Long, Long), (Long, Long), (Long, Long)] { // 重点说明,这里的 firstBroadcastStateDesc,secondBroadcastStateDesc 其实和之前广播流的那两个MapStateDescriptor无关。 // 这里两个MapStateDescriptor是为了存取BroadcastState,这样在 processBroadcastElement和processElement之间就可以传递变量了。我们完全可以定义新的MapStateDescriptor,只要processBroadcastElement和processElement之间认可就行。 // 这里参数 "broadcast-state-1" 是name, flink就是用这个 name 来从Flink运行时系统中存取MapStateDescriptor lazy val firstBroadcastStateDesc = new MapStateDescriptor[Long, Long]( "broadcast-state-1", BasicTypeInfo.LONG_TYPE_INFO.asInstanceOf[TypeInformation[Long]], BasicTypeInfo.LONG_TYPE_INFO.asInstanceOf[TypeInformation[Long]]) val secondBroadcastStateDesc = new MapStateDescriptor[String, String]( "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO) override def processElement( value: (Long, Long), ctx: KeyedBroadcastProcessFunction [Long, (Long, Long), (Long, Long), (Long, Long)]#ReadOnlyContext, out: Collector[(Long, Long)]): Unit = { // 这里Flink源码中是直接把接受到的业务变量直接再次转发出去 out.collect(value) } override def processBroadcastElement( value: (Long, Long), ctx: KeyedBroadcastProcessFunction [Long, (Long, Long), (Long, Long), (Long, Long)]#Context, out: Collector[(Long, Long)]): Unit = { // 这里是把最新传来的广播变量存储起来,processElement中可以取出再次使用. 具体是通过firstBroadcastStateDesc 的 name 来获取 BroadcastState ctx.getBroadcastState(firstBroadcastStateDesc).put(value._1, value._2) ctx.getBroadcastState(secondBroadcastStateDesc).put(value._1.toString, value._2.toString) } } // 广播流和数据流的Source private class CheckpointedSource(val numElements: Int) extends SourceFunction[(Long, Long)] with CheckpointedFunction { private var isRunning = true private var state: ListState[CustomCaseClass] = _ // 就是简单的定期发送 override def run(ctx: SourceFunction.SourceContext[(Long, Long)]) { ctx.emitWatermark(new Watermark(0)) ctx.getCheckpointLock synchronized { var i = 0 while (i < numElements) { ctx.collect(i, i) i += 1 } } // don't emit a final watermark so that we don't trigger the registered event-time // timers while (isRunning) Thread.sleep(20) } } ``` ### 2. 技术难点 #### MapStateDescriptor 首先要说明一些概念: - Flink中包含两种基础的状态:Keyed State和Operator State。 - Keyed State和Operator State又可以 以两种形式存在:原始状态和托管状态。 - 托管状态是由Flink框架管理的状态,如ValueState, ListState, MapState等。 - raw state即原始状态,由用户自行管理状态具体的数据结构,框架在做checkpoint的时候,使用byte[]来读写状态内容,对其内部数据结构一无所知。 - MapState是托管状态的一种:即状态值为一个map。用户通过`put`或`putAll`方法添加元素。 回到我们的例子,广播变量就是OperatorState的一部分,是以托管状态的MapState形式保存的。具体getBroadcastState函数就是DefaultOperatorStateBackend中的实现 所以我们需要用MapStateDescriptor描述broadcast state,这里MapStateDescriptor的使用比较灵活,因为是key,value类似使用,所以个人觉得value直接使用类,这样更方便,尤其是对于从其他语言转到scala的同学。 #### processBroadcastElement ```scala // 因为主要起到控制作用,所以这个函数的处理相对简单 override def processBroadcastElement(): Unit = { // 这里可以把最新传来的广播变量存储起来,processElement中可以取出再次使用,比如 ctx.getBroadcastState(firstBroadcastStateDesc).put(value._1, value._2) } ``` #### processElement ```java // 这个函数需要和processBroadcastElement配合起来使用 override def processElement(): Unit = { // 可以取出processBroadcastElement之前存储的广播变量,然后用此来处理业务变量,比如 val secondBroadcastStateDesc = new MapStateDescriptor[String, String]( "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO) var actualSecondState = Map[String, String]() for (entry <- ctx.getBroadcastState(secondBroadcastStateDesc).immutableEntries()) { val v = secondExpectedBroadcastState.get(entry.getKey).get actualSecondState += (entry.getKey -> entry.getValue) } // 甚至这里只要和processBroadcastElement一起关联好,可以存储任意类型的变量。不必须要和广播变量的类型一致。重点是声明新的对应的MapStateDescriptor // MapStateDescriptor继承了StateDescriptor,其中state为MapState类型,value为Map类型 } ``` #### 结合起来使用 因为某些限制,所以下面只能从网上找一个例子给大家讲讲。 ```java // 模式始终存储在MapState中,并将null作为键。broadcast state始终表示为MapState,这是Flink提供的最通用的状态原语。 MapStateDescriptor

加载全部内容

相关教程
猜你喜欢
用户评论