[大数据] Flink累加器的使用案例

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 大数据 -> Flink累加器的使用案例 -> 正文阅读

[大数据]Flink累加器的使用案例

1、使用累加器，计算输入单词的个数，在job结束后，输出结果

2、代码实现

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

        DataStreamSource<String> line = env.socketTextStream("localhost", 8888);

        line.keyBy(e->e.split(",")[0]).process(
                new KeyedProcessFunction<String, String, Tuple2<String,Integer>>() {

                    private ValueState<Integer> valueState;

                    // 创建一个累加器对象
                    private IntCounter wordCount = new IntCounter();

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("count_state", Integer.class);
                        // Managed Keyed State 方式 所有类型的状态均可以使用
                        stateDescriptor.setQueryable("query_count");
                        valueState = getRuntimeContext().getState(stateDescriptor);
                        System.out.println("job_id=>"+getRuntimeContext().getJobId());

                        // 注册累加器对象
                        getRuntimeContext().addAccumulator("word_count",wordCount);
                    }

                    @Override
                    public void processElement(String line, KeyedProcessFunction<String, String, Tuple2<String, Integer>>.Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
                        String[] fields = line.split(",");
                        String name = fields[0];
                        Integer count = Integer.valueOf(fields[1]);

                        //每处理一个元素，累加器+1
                        wordCount.add(1);

                        Integer oldResult = valueState.value();
                        if(oldResult == null)
                        {
                            oldResult = 0;
                        }

                        count = oldResult + count;
                        valueState.update(count);
                        out.collect(new Tuple2<>(name,count));
                    }
                }
        ).print();

        JobExecutionResult result = env.execute();

        Integer num = result.getAccumulatorResult("word_count");

        System.out.println("计算单词个数是："+num);
    }
}