在哪里可以找到broadcast_state指南中的示例源代码

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class BroadcastStateExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建一个广播状态描述符
        MapStateDescriptor<String, Integer> descriptor =
                new MapStateDescriptor<>("broadcast-state", String.class, Integer.class);

        // 创建一个广播流
        BroadcastStream<Tuple2<String, Integer>> broadcastStream = env.fromElements(
                Tuple2.of("key1", 1),
                Tuple2.of("key2", 2)
        ).broadcast(descriptor);

        // 主数据流
        env.fromElements(
                Tuple2.of("key1", "value1"),
                Tuple2.of("key2", "value2"),
                Tuple2.of("key3", "value3")
        ).flatMap(new MyFlatMapFunction(descriptor)).print();

        env.execute("Broadcast State Example");
    }

    public static class MyFlatMapFunction extends RichFlatMapFunction<Tuple2<String, String>, String> {
        private final MapStateDescriptor<String, Integer> descriptor;
        private transient BroadcastState<String, Integer> broadcastState;

        public MyFlatMapFunction(MapStateDescriptor<String, Integer> descriptor) {
            this.descriptor = descriptor;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            // 获取广播状态
            broadcastState = getRuntimeContext().getBroadcastState(descriptor);
        }

        @Override
        public void flatMap(Tuple2<String, String> value, Collector<String> out) throws Exception {
            String key = value.f0;
            String data = value.f1;

            // 从广播状态中获取数据
            Integer broadcastValue = broadcastState.get(key);

            if (broadcastValue != null) {
                out.collect("Key: " + key + ", Data: " + data + ", Broadcast Value: " + broadcastValue);
            }
        }
    }
}