Getting started with Flink - word statistics

Using batch processing wc

/*
*  The batch 
* */
public class WordCount {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSource<String> data = env.readTextFile("input/a.txt");

        FlatMapOperator<String, Tuple2<String, Long>> wordsTuple = data.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
            String[] words = line.split(" ");
            for (String word : words) {
                out.collect(Tuple2.of(word, 1L));
            }
        }).returns(Types.TUPLE(Types.STRING,Types.LONG));

        //  One thing to note here is the use of Java Medium lambda When the expression , Need to add .returns(Types.TUPLE(Types.STRING,Types.LONG)), Because when Lambda Expressions use  java  When generics ,  Due to the existence of generic erasure ,  Declaration type information to be displayed 

        UnsortedGrouping<Tuple2<String, Long>>  groupWord = wordsTuple.groupBy(0);

        AggregateOperator<Tuple2<String, Long>> result = groupWord.sum(1);

        result.print();
    }
}

Using bounded flow processing wc

/*
*  Bounded flow processing 
* */
public class Word {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);

        DataStream<String> source = environment.readTextFile("input/a.txt");

        SingleOutputStreamOperator<String> flatMap = source.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String line, Collector<String> out) throws Exception {
                String[] words = line.split(" ");
                for (String word : words) {
                    out.collect(word);
                }
            }
        });
        
        // SingleOutputStreamOperator<Tuple2<String, Integer>> map = flatMap.map(word -> Tuple2.of(word, 1)).returns(Types.TUPLE(Types.STRING,Types.INT));

        SingleOutputStreamOperator<Tuple2<String, Integer>> map = flatMap.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
                return Tuple2.of(s, 1);
            }
        });


        KeyedStream<Tuple2<String, Integer>, Tuple> keyedStream = map.keyBy(0);

        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = keyedStream.sum(1);

        sum.print();

        environment.execute();
    }

}

/*
*  Bounded flow processing 
* */
public class Word {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);

        DataStream<String> source = environment.readTextFile("input/a.txt");

        SingleOutputStreamOperator<Tuple2<String, Integer>> result = source.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = line.split(" ");
                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }
            }
        }).keyBy(0).sum(1);

        result.print();


        environment.execute();
    }

}