Flink 广播变量实例

╰+攻爆jí腚メ 2021-09-24 11:38 735阅读 0赞

[https://juejin.im/post/6844903785857286157][https_juejin.im_post_6844903785857286157]

# Flink广播变量简介 #

版本： flink1.9.2,java1.8

### 广播变量用于DataSet： ###

package BroadCast;
    
    import org.apache.flink.api.common.functions.RichMapFunction;
    import org.apache.flink.api.java.DataSet;
    import org.apache.flink.api.java.ExecutionEnvironment;
    import org.apache.flink.configuration.Configuration;
    
    import java.util.ArrayList;
    import java.util.List;
    
    /**
     * @Author you guess
     * @Date 2020/6/24 21:02
     * @Version 1.0
     * @Desc
     */
    public class BroadCastSetTest1 {
    
        public static void main(String[] args) throws Exception {
    
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    
            // 1. 准备等待广播的DataSet数据
            DataSet<Integer> broadCastSet = env.fromElements(300, 400);
    
            DataSet<String> data = env.fromElements("a", "b");
    
            data.map(new RichMapFunction<String, String>() {
    
                //需要定义在map方法里面，不然报错。是匿名内部类
                List<Integer> broadcastList = new ArrayList<>();
    
                @Override
                public void open(Configuration parameters) throws Exception {
                    // 3. 获取广播的DataSet数据 作为一个Collection
                    broadcastList = getRuntimeContext().getBroadcastVariable("name1");
                }
    
                @Override
                public String map(String value) throws Exception {
                    return value + broadcastList.get(value.charAt(0) - "a".charAt(0));
                }
            }).withBroadcastSet(broadCastSet, "name1").print(); // 2. 广播DataSet
    
            //处理数据集 不必写execute，不然有异常
            //env.execute("Flink  BroadCastStreamTest1 ");
        }
    }
    /*
    输出：
    a300
    b400
    */

### 源码：实际用的HashMap存储的 ###

@Public
    public abstract class SingleInputUdfOperator<IN, OUT, O extends SingleInputUdfOperator<IN, OUT, O>>
    	extends SingleInputOperator<IN, OUT, O> implements UdfOperator<O> {
    ......
    
    	@Override
    	public O withBroadcastSet(DataSet<?> data, String name) {
    		if (data == null) {
    			throw new IllegalArgumentException("Broadcast variable data must not be null.");
    		}
    		if (name == null) {
    			throw new IllegalArgumentException("Broadcast variable name must not be null.");
    		}
    
    		if (this.broadcastVariables == null) {
    			this.broadcastVariables = new HashMap<String, DataSet<?>>();
    		}
    
    		this.broadcastVariables.put(name, data);
    
    		@SuppressWarnings("unchecked")
    		O returnType = (O) this;
    		return returnType;
    	}
    ......
    }

## ##

### 广播变量用于DataStream： ###

DataStream<Long> num = source1.broadcast().map(new MapFunction<Long, Long>() {
                @Override
                public Long map(Long value) throws Exception {
                    long id = Thread.currentThread().getId();
                    System.out.println("线程id：" + id + ",接收到数据：" + value);
                    return value;
                }
            });

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAwMDIxODQ_size_16_color_FFFFFF_t_70][]

[https_juejin.im_post_6844903785857286157]: https://juejin.im/post/6844903785857286157
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAwMDIxODQ_size_16_color_FFFFFF_t_70]: /images/20210923/71de38ca65e54f79864a5750bf8b8404.png