Flink Dataset Api（六）广播变量

柔光的暖阳◎ 2023-05-31 10:05 5阅读 0赞

Flink支持广播变量，就是将数据广播到具体的taskmanager上，数据存储在内存中，这样可以减缓大量的shuffle操作；

比如在数据join阶段，不可避免的就是大量的shuffle操作，我们可以把其中一个dataSet广播出去，一直加载到taskManager的内存中，可以直接在内存中拿数据，避免了大量的shuffle，导致集群性能下降；

**注意**：**因为广播变量是要把dataset广播到内存中，所以广播的数据量不能太大，否则会出现OOM这样的问题**

Broadcast：Broadcast是通过withBroadcastSet(dataset，string)来注册的
    
    Access：通过getRuntimeContext().getBroadcastVariable(String)访问广播变量

import org.apache.flink.api.common.functions.RichMapFunction
    import org.apache.flink.configuration.Configuration
    import scala.collection.mutable
    import scala.collection.mutable.{ArrayBuffer,}
    import scala.util.Random
    
    object SourceTest {
      def main(args: Array[String]): Unit = {
        import org.apache.flink.api.scala.extensions._
        import org.apache.flink.api.scala._
        import org.apache.flink.streaming.api.scala.extensions._
    
        val env = ExecutionEnvironment.getExecutionEnvironment
    
        //TODO data2  join  data3的数据，使用广播变量完成
        val data2 = new mutable.MutableList[(Int, Long, String)]
        data2.+=((1, 1L, "Hi"))
        data2.+=((2, 2L, "Hello"))
        data2.+=((3, 2L, "Hello world"))
        val ds1 = env.fromCollection(Random.shuffle(data2))
        val data3 = new mutable.MutableList[(Int, Long, Int, String, Long)]
        data3.+=((1, 1L, 0, "Hallo", 1L))
        data3.+=((2, 2L, 1, "Hallo Welt", 2L))
        data3.+=((2, 3L, 2, "Hallo Welt wie", 1L))
        val ds2 = env.fromCollection(Random.shuffle(data3))
    
        //todo 使用内部类RichMapFunction，提供open和map，可以完成join的操作
        val result = ds1.map(new RichMapFunction[(Int , Long , String) , 
                        ArrayBuffer[(Int , Long , String , String)]] {
    
          var brodCast:mutable.Buffer[(Int, Long, Int, String, Long)] = null
    
          override def open(parameters: Configuration): Unit = {
            import scala.collection.JavaConverters._
            //asScala需要使用隐式转换
            brodCast = this.getRuntimeContext.getBroadcastVariable[(Int, Long, Int, String, 
                        Long)]("ds2").asScala
          }
          override def map(value: (Int, Long, String)):ArrayBuffer[(Int , Long , String , String)] = {
            val toArray: Array[(Int, Long, Int, String, Long)] = brodCast.toArray
            val array = new mutable.ArrayBuffer[(Int , Long , String , String)]
            var index = 0
    
            var a:(Int, Long, String, String) = null
            while(index < toArray.size){
              if(value._2 == toArray(index)._5){
                a = (value._1 , value._2 , value._3 , toArray(index)._4)
                array += a
              }
              index = index + 1
            }
            array
          }
        }).withBroadcastSet(ds2 , "ds2")
        println(result.collect())
      }
    }