Flink 广播流使用

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// ...env setting

// Source
KafkaSource<SourceValue> kafkaSource =
    KafkaSource.<>builder()
        .setBootstrapServers(BOOTSTRAP_SERVERS)
        .setTopics(TOPIC)
        .build();
DataStreamSource<SourceValue> source =
    env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "sourceName");

// Broadcast
MapStateDescriptor<Key, Value> state =
    new MapStateDescriptor<>("state", Key.class, Value.class);
BroadcastStream<Value> broadcastStream = env.addSource().broadcast(state);

// Process
source.connect(broadcastStream)
    .process(new MyProcessFunction())
    .addSink(new MySinkFunction());

env.execute();