当前位置：首页 > article >正文

Flink 实现无界流

article 2024/11/15 11:40:12

Flink 实现无界流

package org.example.test;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * DataSet API使用
 */
public class WordCount2 {

    public static void main(String[] args) throws Exception {
        //该类主要是用于进行批处理
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //读取文本
        DataSource<String> stringDataStreamSource = env.readTextFile("input/test.txt");
        //进行ETL处理,Tuple2 是二元数组的意思
        FlatMapOperator<String, Tuple2<String, Integer>> stringTuple2FlatMapOperator = stringDataStreamSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = value.split(" ");
                for (String word : words) {
                    Tuple2<String, Integer> oneTuple2 = Tuple2.of(word, 1);
                    out.collect(oneTuple2);
                }
            }
        });
        //进行分组，分组字段取下标第0个
        UnsortedGrouping<Tuple2<String, Integer>> tuple2UnsortedGrouping =
                stringTuple2FlatMapOperator.groupBy(0);
        //进行sum操作
        AggregateOperator<Tuple2<String, Integer>> sum = tuple2UnsortedGrouping.sum(1);
        sum.print();
    }
}