在Flink中，同一個運算元可能存在若干個不同的並行例項，計算過程可能不在同一個Slot中進行，不同運算元之間更是如此，因此不同運算元的計算資料之間不能像Java陣列之間一樣互相訪問，而廣播變數Broadcast便是解決這種情況的。如下程式碼所示：

    val env = ExecutionEnvironment.getExecutionEnvironment
    val ds1 = env.fromElements("1", "2", "3", "4", "5")
    val ds2 = env.fromElements("a", "b", "c", "d", "e")

    ds1.map
 
{
      t =>
        (t, ds2)
    }.print()

執行上述程式碼會報InvalidProgramException的錯，因為在ds1的map運算元中無法再去呼叫ds2，此時可以使用廣播變數將ds2這個變數進行廣播，使得ds2這一被廣播的資料集在ds1中map運算元的所有並行例項中都可用，具體處理方式如下：

object BroadcastTest {
  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment
    val ds1 = env.fromElements("1"
 
, "2", "3", "4", "5")
    val ds2 = env.fromElements("a", "b", "c", "d", "e")

    ds1.map(new RichMapFunction[String, (String, String)] {
      private var ds2: Traversable[String] = null

      override def open(parameters: Configuration) {
        ds2 = getRuntimeContext.getBroadcastVariable[String
 
]("broadCast").asScala
      }

      def map(t: String): (String, String) = {
        var result = ""
        for (broadVariable <- ds2) {
          result = result + broadVariable + " "
        }
        (t, result)
      }
    }).withBroadcastSet(ds2, "broadCast").print()
  }
}

上述程式碼能夠成功執行出如下結果：

(1,a b c d e )
(2,a b c d e )
(3,a b c d e )
(4,a b c d e )
(5,a b c d e )

可以看到，ds1的map運算元通過訪問廣播變數成功訪問到ds2中的資料。該過程分為兩步：設定廣播變數和獲取廣播變數。

1. 設定廣播變數
   在某個需要用到該廣播變數的運算元後呼叫withBroadcastSet(var1, var2)進行設定，var1為需要廣播變數的變數名，var2是自定義變數名，為String型別。注意，被廣播的變數只能為DataSet型別，不能為List、Int、String等型別。
2. 獲取廣播變數
   建立該運算元對應的富函式類，例如map函式的富函式類是RichMapFunction，該類有兩個構造引數，第一個引數為運算元輸入資料型別，第二個引數為運算元輸出資料型別。首先建立一個Traversable[_]介面用於接收廣播變數並初始化為空，接收型別與運算元輸入資料型別相對應；然後重寫open函式，通過getRuntimeContext.getBroadcastVariable[_](var)獲取到廣播變數，var即為設定廣播變數時的自定義變數名，型別為String，open函式在運算元生命週期的初始化階段便會呼叫；最後在map方法中對獲取到的廣播變數進行訪問及其它操作。

注意：只有在某個Operator中使用到不屬於該Operator的DataSet時才需要廣播變數，在iterate內部可以將某個DataSet直接作為起始節點，不需要使用廣播變數。