赞
踩
设置广播变量
在某个需要用到该广播变量的算子后调用withBroadcastSet(var1, var2)进行设置,var1为需要广播变量
的变量名,var2是自定义变量名,为String类型。注意,被广播的变量只能为DataSet类型,不能为
List、Int、String等类型。
获取广播变量
创建该算子对应的富函数类,例如map函数的富函数类是RichMapFunction,该类有两个构造参数,第
一个参数为算子输入数据类型,第二个参数为算子输出数据类型。首先创建一个Traversable[]接口用于
接收广播变量并初始化为空,接收类型与算子输入数据类型相对应;然后重写open函数,通过
getRuntimeContext.getBroadcastVariable获取到广播变量,var即为设置广播变量时的自定义变量
名,类型为String,open函数在算子生命周期的初始化阶段便会调用;最后在map方法中对获取到的广
播变量进行访问及其它操作。
注意:只有在某个Operator中使用到不属于该Operator的DataSet时才需要广播变量,在iterate内部可
以将某个DataSet直接作为起始节点,不需要使用广播变量
再次我写了两个例子
下面附上Scala代码
object DataSetBroadcastApp { //TODO 广播变量 def main(args: Array[String]): Unit = { val env = ExecutionEnvironment.getExecutionEnvironment broadcast(env) } def broadcast(env:ExecutionEnvironment): Unit ={ //TODO 创建两个元素 val ds1 = env.fromElements("1", "2", "3", "4", "5") val ds2 = env.fromElements("a", "b", "c", "d", "e") //TODO 广播数据集 ds1.map(new RichMapFunction[String,(String,String)] { //TODO 定义一个私有的另一个数据集 private var ds2 : Traversable[String] = null //TODO 因为是Java的集合所以要用隐式转换 import scala.collection.JavaConverters._ //TODO 参见官网https://ci.apache.org/projects/flink/flink-docs-release- 1.10/dev/batch/#broadcast-variables override def open(parameters: Configuration): Unit = { ds2 = getRuntimeContext.getBroadcastVariable[String]("broadcast").asScala } override def map(value: String): (String, String) = { //TODO 定义一个结果 var result = " " for(broadVariable <- ds2){ result = result + broadVariable + " " } (value,result) } }).withBroadcastSet(ds2,"broadcast").print() } }
Java代码
public class JavaDataSetBroadcastApp { //TODO Java实现广播变量 public static void main(String[] args) throws Exception{ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); broadcast(env); } public static void broadcast(ExecutionEnvironment env) throws Exception { DataSet<Integer> toBroadcast = env.fromElements(1, 2, 3); DataSet<String> data = env.fromElements("a", "b"); MapOperator<String, String> broadcastSet = data.map( new MyMapperFunction() ).withBroadcastSet( toBroadcast, "huan" ); broadcastSet.printToErr(); } } class MyMapperFunction extends RichMapFunction<String, String> { //TODO 定义一个私有集合 private List list= new ArrayList(); @Override public void open(Configuration parameters) throws Exception { //TODO 作为集合访问广播数据集.所以要定义集合,用集合添加广播变量,因为返回值是集 合,所以要用集合去添加广播变量 Collection<Integer> broadcastSet = getRuntimeContext().getBroadcastVariable( "huan" ); list.add( broadcastSet ); } @Override public String map(String value) throws Exception { //TODO value的值就是data list的值就是toBroadcast return value+ " : " +list; } }
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。