Flink / Scala - DataStream Broadcast State 模式示例详解-阿里云开发者社区

一.引言

上一篇文章 Flink / Scala - DataSet 应用 Broadcast Variables 介绍了 DataSet 场景下 Broadcast 的使用，本文将介绍 DataStream 中的 Broadcast 应用场景，与 DataSet 类似，Broadcast 的值是所有 task 公用的，Broadcast State 是为 DataStreaming 所有 task 定制的可实时修改的公用值。

二.代码常规介绍

DataStream<T> output = dataStream
                 .connect(BroadcastStream)
                 .process(
                     // KeyedBroadcastProcessFunction 中的类型参数表示：
                     //   1. key stream 中的 key 类型
                     //   2. 非广播流中的元素类型
                     //   3. 广播流中的元素类型
                     //   4. 结果的类型，在这里是 string
                     new KeyedBroadcastProcessFunction<Ks, In1, In2, Out>() {
                         // 模式匹配逻辑
                     }
                 );

常规使用中我们都包含一个数据流 DataStream，其中包含我们需要处理的数据，如果处理逻辑会随着一个状态值的改变而改变，这是可以引入第二个数据流成为广播流 BroadcastStream，通过调用 DataStream 的 connect 方法，并将 BroadcastStream 参数传入即可获得一个 BroadcastConnectedStream，这时数据同时包含数据流和状态流，需要重写 process 函数处理两个流的数据，根据 DataStream 是否是 Keyd-Stream，Process 方法分为:

· keyed 流，那就是 KeyedBroadcastProcessFunction 类型

· non-keyed 流，那就是 BroadcastProcessFunction 类型

在传入的 BroadcastProcessFunction 或 KeyedBroadcastProcessFunction 中，我们需要实现两个方法。processBroadcastElement() 方法负责处理广播流中的元素，processElement() 负责处理非广播流中的元素。两个子类型定义如下：

public abstract class BroadcastProcessFunction<IN1, IN2, OUT> extends BaseBroadcastProcessFunction {
    public abstract void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception;
    public abstract void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception;
}
public abstract class KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT> {
    public abstract void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception;
    public abstract void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception;
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception;
}

需要注意的是 processBroadcastElement() 负责处理广播流的元素，而 processElement() 负责处理另一个流的元素。两个方法的第二个参数(Context)不同，均有以下方法：

得到广播流的存储状态：ctx.getBroadcastState(MapStateDescriptor<K, V> stateDescriptor)

查询元素的时间戳：ctx.timestamp()

查询目前的 Watermark：ctx.currentWatermark()

目前的处理时间 (processing time)：ctx.currentProcessingTime()

产生旁路输出：ctx.output(OutputTag<X> outputTag, X value)

三.应用实例

上面说的比较官方，下面通过一个简单的例子理解一下 BroadCast Value 和 BroadCast Stream 的用处，上面提到了 BroadCast Stream 作为一个状态流控制 DataStream 的数据输出，下面实现以下功能：

DataStream: 定期生成 num - 100+num 的 100 个数字，每次生成周期初始化 num + 100

BroadCastStream：不定期传入状态控制输出状态，分为 odd-单数 even-双数

Sink：根据 odd 和 even 的状态，print 输出 100 个数字中的单数或者双数

1.DataStream

5s 中生成 num - (num+100) 的数字，下一批数据比上一批增加 100，这里继承 RichSourceFunction 自定义 Source 来源然后通过 addSource 实现，完整的 DataStream Source 生成方法参考: Flink / Scala - DataSource 之 DataStream 获取数据总结。

// 每5s生成一批数据 数据流
    case class InputData(num: Int)
    class SourceFromCollection extends RichSourceFunction[InputData] {
      private var isRunning = true
      var start = 0
      override def run(ctx: SourceFunction.SourceContext[InputData]): Unit = {
        while ( {
          isRunning
        }) {
          (start to (start + 100)).foreach(num => {
            ctx.collect(InputData(num))
          })
          start += 100
          TimeUnit.SECONDS.sleep(5)
        }
      }
      override def cancel(): Unit = {
        isRunning = false
      }
    }
    val keyedStream = env.addSource(new SourceFromCollection()).setParallelism(1).keyBy(_.num)

上述流根据 num- num+100 的数字生成 InputData 类，并通过 keyBy 生成 Keyd-Stream。

2.BroadCastStream

BroadCastStream 广播流即本例中的状态流，这里通过 File 传递状态值并解析，同样是继承 RichFunction 实现自定义的 Source，每 1s 从对应文件读取，获取是否有新的状态传入。

// MapStateDescriptor odd: 奇数 even: 偶数
    case class FilterState(state: String)    
    // 每s监控一次文件，并读取最新的状态
    class SourceFromFile extends RichSourceFunction[String] {
      private var isRunning = true
      override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
        val bufferedReader = new BufferedReader(new FileReader("./data.txt"))
        while ( {
          isRunning
        }) {
          val line = bufferedReader.readLine
          if (!StringUtils.isBlank(line)) {
            ctx.collect(line)
          }
          TimeUnit.SECONDS.sleep(1)
        }
      }
      override def cancel(): Unit = {
        isRunning = false
      }
    }
    val ruleStateDescriptor = new MapStateDescriptor("RulesBroadcastState", classOf[String], classOf[FilterState])
    // 广播流，广播规则并且创建 BroadCast
    val ruleStream = env.addSource(new SourceFromFile).setParallelism(1).map(new RichMapFunction[String, FilterState]() {
      override def map(in: String): FilterState = {
        FilterState(in)
      }
    }).broadcast(ruleStateDescriptor)

stateDescriptor 负责声明广播状态的类型，这里定义为 MapStateDescriptor ，后续通过 String 类型的 key 即可获取对应的 FilterState，从而决定 DataStream 中的数据如何 sink。

3.合并 DataStream 与 BroadCastStream

DataStream.connect(BroadCastStream)，由于原始 DataStream 为 keyd-stream，所以使用 keyedBroadcastProcessFunction，共包含四个参数:

· ks - keyBy 字段的类型，这里根据 InputData.num keyBy，所以是 Int

· IN1 - DataStream 数据流的类型，这里是 InputData

· IN2 - BroadCastStream 广播流的类型，这里是 FilterState

· OUT - Sink 输出端为直接输出 Print String，所以为 String

keyedStream.connect(ruleStream).process(new KeyedBroadcastProcessFunction[Int, InputData, FilterState, String] {
      // 与之前的 Descriptor 相同
      val ruleStateDescriptor = new MapStateDescriptor("RulesBroadcastState", classOf[String], classOf[FilterState])
      override def processElement(inputData: InputData, context: KeyedBroadcastProcessFunction[Int, InputData, FilterState, String]#ReadOnlyContext, out: Collector[String]): Unit = {
        val filterStateClass = context.getBroadcastState(ruleStateDescriptor).get("broadcastStateKey")
        val filterState = if (filterStateClass == null) {
          "odd"
        } else {
          filterStateClass.state
        }
        // 奇数模式
        if (filterState == "odd" && inputData.num % 2 != 0) {
          out.collect(inputData.num.toString)
        }
        // 偶数模式
        if (filterState == "even" && inputData.num % 2 == 0) {
          out.collect(inputData.num.toString)
        }
      }
      override def processBroadcastElement(filterState: FilterState, context: KeyedBroadcastProcessFunction[Int, InputData, FilterState, String]#Context, collector: Collector[String]): Unit = {
        // 从广播中获取规则
        val broadCastValue = context.getBroadcastState(ruleStateDescriptor)
        broadCastValue.put("broadcastStateKey", filterState)
        println(s"Rule Changed: ${filterState.state}")
      }
    }).setParallelism(1).print()

A. ProcessElement

该方法负责输出数据，根据 FilterState 的状态是 odd-单数还是 even-双数，状态默认为 odd-单数。通过 context.getBroadcastState(StateDescriptor) 方法获取 BroadcastStream 中的 FilterState 数据。注意这里的 StateDescriptor 要与上面初始化的 StateDescriptor 保持一致。

B. ProcessBroadcastElement

该方法负责处理 Broadcast 数据流并更新至 context，从而其他 task 节点在执行 processElement 方法时获取最新的状态值，这里 put 的 Key 和上述方法 get 的 Key 需要保持一致，否则获取状态值为 null。

4. 测试

为了本地测试方便查看，两个 Stream 的 parallelism 都设置为1。

状态文件 File 为空，此时默认状态为 odd，输出单数：

编辑

文件内增加一行 even，并 ctrl s 保存，此时 Broadcast 1s 的间隔检测到新状态 even，处理并更细至各 task，各 task 输出偶数：

编辑

再次增加一行 odd，此时输出状态改变，重新修改为输出单数：

编辑

一个基本的 BroadcastValue 控制 DataStream 的实例就完成了，状态文件夹最终包含两行状态数据:

编辑

5.完整代码

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
import org.apache.commons.lang3.StringUtils
import java.io.BufferedReader
import java.io.FileReader
import java.util.concurrent.TimeUnit
object BroadCastStateDemo {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 每5s生成一批数据 数据流
    case class InputData(num: Int)
    class SourceFromCollection extends RichSourceFunction[InputData] {
      private var isRunning = true
      var start = 0
      override def run(ctx: SourceFunction.SourceContext[InputData]): Unit = {
        while ( {
          isRunning
        }) {
          (start to (start + 100)).foreach(num => {
            ctx.collect(InputData(num))
          })
          start += 100
          TimeUnit.SECONDS.sleep(5)
        }
      }
      override def cancel(): Unit = {
        isRunning = false
      }
    }
    val keyedStream = env.addSource(new SourceFromCollection()).setParallelism(1).keyBy(_.num)
    // 每s监控一次文件，并读取最新的状态
    class SourceFromFile extends RichSourceFunction[String] {
      private var isRunning = true
      override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
        val bufferedReader = new BufferedReader(new FileReader("/Users/xudong11/flink/src/main/scala/com.weibo.ug.push.flink/DataStreamingDemo/data.txt"))
        while ( {
          isRunning
        }) {
          val line = bufferedReader.readLine
          if (!StringUtils.isBlank(line)) {
            ctx.collect(line)
          }
          TimeUnit.SECONDS.sleep(1)
        }
      }
      override def cancel(): Unit = {
        isRunning = false
      }
    }
    // MapStateDescriptor odd: 奇数 even: 偶数
    case class FilterState(state: String)
    val ruleStateDescriptor = new MapStateDescriptor("RulesBroadcastState", classOf[String], classOf[FilterState])
    // 广播流，广播规则并且创建 BroadCast
    val ruleStream = env.addSource(new SourceFromFile).setParallelism(1).map(new RichMapFunction[String, FilterState]() {
      override def map(in: String): FilterState = {
        FilterState(in)
      }
    }).broadcast(ruleStateDescriptor)
    // 连接两个流
    keyedStream.connect(ruleStream).process(new KeyedBroadcastProcessFunction[Int, InputData, FilterState, String] {
      // 与之前的 Descriptor 相同
      val ruleStateDescriptor = new MapStateDescriptor("RulesBroadcastState", classOf[String], classOf[FilterState])
      override def processElement(inputData: InputData, context: KeyedBroadcastProcessFunction[Int, InputData, FilterState, String]#ReadOnlyContext, out: Collector[String]): Unit = {
        val filterStateClass = context.getBroadcastState(ruleStateDescriptor).get("broadcastStateKey")
        val filterState = if (filterStateClass == null) {
          "odd"
        } else {
          filterStateClass.state
        }
        // 奇数模式
        if (filterState == "odd" && inputData.num % 2 != 0) {
          out.collect(inputData.num.toString)
        }
        // 偶数模式
        if (filterState == "even" && inputData.num % 2 == 0) {
          out.collect(inputData.num.toString)
        }
      }
      override def processBroadcastElement(filterState: FilterState, context: KeyedBroadcastProcessFunction[Int, InputData, FilterState, String]#Context, collector: Collector[String]): Unit = {
        // 从广播中获取规则
        val broadCastValue = context.getBroadcastState(ruleStateDescriptor)
        broadCastValue.put("broadcastStateKey", filterState)
        println(s"Rule Changed: ${filterState.state}")
      }
    }).setParallelism(1).print()
    env.execute()
  }
}

四.总结

1.实现步骤

Broadcast Value 通过 DataStream connect BroadCastStream 连接实现，期间注意两个 ProcessFunction 的重写与对应 StateDescriptor 的定制即可。

2.数据一致性

其次需要注意两个 processFunction 的参数 ctx，在 processElement 中 ctr 是 readOnly，因为一致性的原因，这里只允许 task 读取最新的 State 但不能修改；相反的 processBroadcastElement 方法中的 context 允许修改其中 value 状态的值，注意这里的逻辑要保持全局的一致性（增加随机数随机修改状态值可视作是不保持全局唯一性的操作），否则会造成状态不同从而导致 task 端输出不一致。

3.CheckPoint

所有的 task 均会对 broadcast state 进行 checkpoint：虽然所有 task 中的 broadcast state 是一致的，但当 checkpoint 来临时所有 task 均会对 broadcast state 做 checkpoint。这个设计是为了防止在作业恢复后读文件造成的文件热点即 hotspot 。当然这种方式会造成 checkpoint 一定程度的写放大，放大倍数为 p（=并行度）。Flink 会保证在恢复状态 / 改变并发的时候数据没有重复且没有缺失。在作业恢复时，如果与之前具有相同或更小的并发度，所有的 task 读取之前已经 checkpoint 过的 state。在增大并发的情况下，task 会读取本身的 state，多出来的并发（p_new - p_old）会使用轮询调度算法读取之前 task 的 state。

4.State Backend

broadcast state 在运行时保存在内存中，需要保证内存充足。这一特性同样适用于所有其他 Operator State，因此不使用 RocksDB state backend。

Flink / Scala - DataStream Broadcast State 模式示例详解

一.引言

二.代码常规介绍

三.应用实例

1.DataStream

2.BroadCastStream

3.合并 DataStream 与 BroadCastStream

4. 测试

5.完整代码

四.总结

1.实现步骤

2.数据一致性

3.CheckPoint

4.State Backend

热门文章

最新文章

相关课程

相关电子书

相关实验场景

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

Flink / Scala - DataStream Broadcast State 模式示例详解

一.引言

二.代码常规介绍

三.应用实例

1.DataStream

2.BroadCastStream

3.合并 DataStream 与 BroadCastStream

4. 测试

5.完整代码

四.总结

1.实现步骤

2.数据一致性

3.CheckPoint

4.State Backend

热门文章

最新文章

相关课程

相关电子书

相关实验场景