Flink 1.17教程：wordcount maven工程java代码示例（批、流实现方式）

Dear 丶 2023-10-15 14:35 32阅读 0赞

> 批、流实现wordcount代码示例

## pom.xml ##

<properties>
            <flink.version>1.17.0</flink.version>
        </properties>
     
        <dependencies>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java</artifactId>
                <version>${flink.version}</version>
            </dependency>
     
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-clients</artifactId>
                <version>${flink.version}</version>
            </dependency>
        </dependencies>

## 代码 ##

### DataSet批处理实现Wordcount ###

package com.atguigu.wc;
     
    import org.apache.flink.api.common.functions.FlatMapFunction;
    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.util.Collector;
     
    /**
     * TODO DataSet API 实现 wordCount
     */
    public class WordCountBatchDemo {
        
        public static void main(String[] args) throws Exception {
        
            // TODO 1.创建执行环境
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
     
            // TODO 2.读取文件：从文件中读取
            DataSource<String> lineDS = env.readTextFile("input/word.txt");
     
            // TODO 3.切分、转换（word, 1）
            FlatMapOperator<String, Tuple2<String, Integer>> wordAndOne = lineDS.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
        
                @Override
                public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
        
                    // TODO 3.1 按照空格切分单词
                    String[] words = value.split(" ");
                    // TODO 3.2 将单词转换为(word, 1)格式
                    for (String word : words) {
        
                        Tuple2<String, Integer> wordTuple2 = Tuple2.of(word, 1);
                        // TODO 3.3 使用Collector向下游发送数据
                        out.collect(wordTuple2);
                    }
                }
            });
     
            // TODO 4.按照word分组
            UnsortedGrouping<Tuple2<String, Integer>> wordAndOneGroupBy = wordAndOne.groupBy(0);
     
            // TODO 5.各分组内聚合
            AggregateOperator<Tuple2<String, Integer>> sum = wordAndOneGroupBy.sum(1); //1是位置，表示第二个元素
     
            // TODO 6.输出
            sum.print();
        }
    }

### ctrl + p：查看传参方式。 ###

![img][]

> ctrl + p：查看传参方式。

![img][img 1]

### src同级根目录：input/word.txt ###

hello flink
    hello world
    hello java

### 执行结果 ###

![img][img 2]

### DataStream有界流实现Wordcount ###

package com.atguigu.wc;
     
    import org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.api.java.functions.KeySelector;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.datastream.KeyedStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.util.Collector;
     
    /**
     * TODO DataStream实现Wordcount：读文件（有界流）
     *
     */
    public class WordCountStreamDemo {
        
        public static void main(String[] args) throws Exception {
        
            // TODO 1.创建执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
     
            // TODO 2.读取数据:从文件读
            DataStreamSource<String> lineDS = env.readTextFile("input/word.txt");
     
            // TODO 3.处理数据: 切分、转换、分组、聚合
            // TODO 3.1 切分、转换
            SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOneDS = lineDS //<输入类型, 输出类型>
                    .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) {
        
                                // 转换成 二元组 （word，1）
                                Tuple2<String, Integer> wordsAndOne = Tuple2.of(word, 1);
                                // 通过 采集器 向下游发送数据
                                out.collect(wordsAndOne);
                            }
                        }
                    });
            // TODO 3.2 分组
            KeyedStream<Tuple2<String, Integer>, String> wordAndOneKS = wordAndOneDS.keyBy(
                    new KeySelector<Tuple2<String, Integer>, String>() {
        
                        @Override
                        public String getKey(Tuple2<String, Integer> value) throws Exception {
        
                            return value.f0;
                        }
                    }
            );
            // TODO 3.3 聚合
            SingleOutputStreamOperator<Tuple2<String, Integer>> sumDS = wordAndOneKS.sum(1);
     
            // TODO 4.输出数据
            sumDS.print();
     
            // TODO 5.执行：类似 sparkstreaming最后 ssc.start()
            env.execute();
        }
    }
     
    /**
     * 接口 A，里面有一个方法a()
     * 1、正常实现接口步骤：
     * <p>
     * 1.1 定义一个class B  实现 接口A、方法a()
     * 1.2 创建B的对象：   B b = new B()
     * <p>
     * <p>
     * 2、接口的匿名实现类：
     * new A(){
     * a(){
     * <p>
     * }
     * }
     */

### 批、流代码对比 ###

![img][img 3]

1.  创建执行环境的不同，流处理程序使用的是 StreamExecutionEnvironment。
2.  转换处理之后，得到的数据对象类型不同。
3.  分组操作调用的是 keyBy 方法，可以传入一个匿名函数作为键选择器（KeySelector）， 指定当前分组的 key 是什么。
4.  代码末尾需要调用 env 的 execute 方法，开始执行任务。

--------------------

links：

[https://blog.csdn.net/weixin\_44949135/article/details/130895033][https_blog.csdn.net_weixin_44949135_article_details_130895033]

[img]: https://img-blog.csdnimg.cn/dca177998aa1423486d7ca570c3025b2.png
[img 1]: https://img-blog.csdnimg.cn/8a16fd539dcf4eeb9c9c244be3f492d3.png
[img 2]: https://img-blog.csdnimg.cn/09539bd942f948e59198a78ecd37e2ba.png
[img 3]: https://img-blog.csdnimg.cn/69bae154d4b049e9aed92581a544b6a7.png
[https_blog.csdn.net_weixin_44949135_article_details_130895033]: https://blog.csdn.net/weixin_44949135/article/details/130895033