flink.11 DataStream api之广播

广播意思就是将变量发送到每一个并行运行的task所在的机器上,这样可以避免数据在涉及到聚合之时的跨网传输,提高流运行的速度,正因为广播是提前将需要的公共数据发送到各个集群的节点上,所以来说,广播不适合广播大量的数据.

构造数据流A, B 这里假设B为要被广播的流数据,A为普通数据流,A需要用B流做一些逻辑运算为广播流构造描述符对象

> MapStateDescriptor ruleStateDescriptor = new
> MapStateDescriptor<>( 			"RulesBroadcastState",
> 			BasicTypeInfo.STRING_TYPE_INFO, 			TypeInformation.of(new
> TypeHint() {}));

env调用广播方法 broadcase()非广播流调用connect(广播流) 返回合并流对象,合并流对象调用 process(自己实现的接口), 接口中定义处理逻辑

DataStream output = colorPartitionedStream
                 .connect(ruleBroadcastStream)
                 .process(
                     
                     // type arguments in our KeyedBroadcastProcessFunction represent: 
                     //   1. the key of the keyed stream
                     //   2. the type of elements in the non-broadcast side
                     //   3. the type of elements in the broadcast side
                     //   4. the type of the result, here a string
                     
                     new KeyedBroadcastProcessFunction() {
                         // my matching logic
                     }
                 );
 

5.编写处理逻辑,处理逻辑有两个接口,BroadcastProcessFunction 和 KeyedBroadcastProcessFunction ,前者用于处理非键控流,后者用于处理键控流. 每个接口都有两个核心的方法:processElement,和processBroadcastElement, processBroadcastElement用于处理接收到的广播流,一般来说会调用ctx.getBroadcastState(mapStateDescriptor);
broadcastState.put(value, value); 意思是从数据源B流中获取新订阅到的广播流数据,然后填充到全局的广播流中. processElement主要用于获取广播流,然后获取数据流,然后定义自己的处理逻辑,处理完了之后发送到下游

 StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
        environment.enableCheckpointing(1000 * 180);
        FlinkKafkaConsumer010 location = KafkaUtil.getConsumer("event_stream", "test_1", "test");
        FlinkKafkaConsumer010 object = KafkaUtil.getConsumer("bro_stream", "test_2", "test");
        // 把事件流按key进行分流，这样相同的key会发到同一个节点
        KeyedStream driverDatastream = environment.addSource(location).map(new MapFunction() {

            @Override
            public People map(String s) throws Exception {
                return parse(s);
            }
        }).keyBy((KeySelector) people -> people.id);
    
        // 描述这个map ,key value都为string 
        MapStateDescriptor mapStateDescriptor = new MapStateDescriptor("register", Types.STRING, Types.STRING);
        BroadcastStream broadcast = environment.addSource(object).broadcast(mapStateDescriptor);
        driverDatastream.connect(broadcast).process(new Patternevaluator()).print();
        try {
            environment.execute("register collect");
        } catch (Exception e) {
            e.printStackTrace();
        }

因为主类中用了键控流,(所谓键控流就是根据key select 对流数据进行分区,相同key的数据会发送到同一个线程中处理),所以要用接口KeyedBroadcastProcessFunction

public class Patternevaluator extends KeyedBroadcastProcessFunction {

    MapStateDescriptor mapStateDescriptor;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 这里需要初始化map state 描述
        mapStateDescriptor = new MapStateDescriptor("register", Types.STRING, Types.STRING);

    }

    // 处理每一个元素，看state是否有匹配的，有的话，下发到下一个节点
    @Override
    public void processElement(People value, ReadonlyContext ctx, Collector out) throws Exception {
        ReadOnlyBroadcastState broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        if ((value.getIdCard() != null && broadcastState.get(value.getIdCard()) != null) || (value.getPhone() != null && broadcastState.get(value.getPhone()) != null)) {
            System.out.println("匹配到" + value.toString());
            out.collect(value);
        }

    }


    // 新增加的广播元素，放入state中
    @Override
    public void processBroadcastElement(String value, Context ctx, Collector out) throws Exception {
        System.out.println("新增加需要监控的" + value.toString());
        BroadcastState broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        broadcastState.put(value, value);
    }
}

代码部分参考:广播
附上官网:官网