Flink从入门到实战一[快速上手]-4-Flink HelloWorld之流处理WordCount

flink中的数据分为有界流数据和无界流数据，对应的处理计算叫做批处理和流处理，批处理对应的API是DataSet，流处理对应的API是DataStream。

Flink1.14.3简单demo

我们接下来以Flink1.14.3版本为例实现一个Flink简单demo，以流处理方式实现WordCount，从文件中读取内容，作为输入源进行计算

1、pom引入

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-java</artifactId>
  <version>1.14.3</version>
</dependency>
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-java_2.11</artifactId>
  <version>1.14.3</version>
</dependency>
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-clients_2.11</artifactId>
  <version>1.14.3</version>
</dependency>

2、代码实现

package org.itzhimei;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.client.program.StreamContextEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 流处理实现WordCount
 */
public class WordCountStream {

    public static void main(String[] args) throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamContextEnvironment.getExecutionEnvironment();

        //从socket流读取数据
        String input = "D:\\______flink______\\input\\wordcount.txt";
        DataStream<String> dataStream = env.readTextFile(input);

        //数据计算
        //通过自定义类MyBatchFlatMapper，实现分词
        //keyBy(item->item.f0)进行分组，取二元组第一个元素进行分组
        //通过sum(1)进行汇总，1标识二元组中的第二个值，也就是单词出现的次数1
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = dataStream.flatMap(new MyStreamFlatMap())
                .keyBy(item->item.f0)
                .sum(1);
        result.print();
        env.execute();
    }

    private static class MyStreamFlatMap implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
            String[] words = s.split(" ");
            for(String word:words) {
                collector.collect(new Tuple2<>(word,1));
            }
        }
    }
}

3、测试数据
hello Flink
hello Java
how are you
I’m fine thank you and you
I’m Ok

4、输出结果

16> (Flink,1)
6> (thank,1)
15> (and,1)
9> (fine,1)
3> (I'm,1)
3> (I'm,2)
10> (you,1)
1> (Ok,1)
5> (hello,1)
5> (hello,2)
10> (you,2)
8> (are,1)
10> (you,3)
13> (Java,1)
11> (how,1)

从输出结果可以看出，每一个单词更新了数据，马上就输出了，并不是像批处理，所有结果只输出一次最终结果，这里的流处理是计算一次输出一次最新结果。
比如“I’m”
3> (I’m,1)
3> (I’m,2)
是连续输出的。

5、API分析

基本流程是：
1）创建执行环境对象
2）获取数据源
3）计算
4）输出结果