Flink：入门

梦里梦外; 2024-03-24 17:09 33阅读 0赞

### 单词统计案例（批数据） ###

**需求**

统计一个文件中各个单词出现的次数，把统计结果输出到文件

**依赖**

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.7.2</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.12</artifactId>
        <version>1.7.2</version>
        <scope>provided</scope>
    </dependency>

**代码**

import org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.api.java.DataSet;
    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;
    
    
    public class WordCountBatch {
    
        public static void main(String[] args) throws Exception {
            // 1、获取flink的运行环境
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    
            // 2、用flink的运行环境，去获取待分析数据
            String input = "C:\\data\\input\\hello.txt";
            String output = "C:\\data\\output";
            DataSource<String> lines = env.readTextFile(input);
    
            // 3、处理数据
            //将lines中的每一行元素都施加一个算法
            FlatMapOperator<String, Tuple2<String, Integer>> wordAndOne = lines.flatMap(new SplitFun());
    
            // (单词，1)
            // b。把相同的单词聚合到一起
            UnsortedGrouping<Tuple2<String, Integer>> groupByWord = wordAndOne.groupBy(0);
            // c。把聚合到一起的数据累加处理
            DataSet<Tuple2<String, Integer>> result = groupByWord.sum(1);
            // 4、保存处理结果
            result.writeAsText(output);
            result.print();
            // 5、触发程序执行
            env.execute("wordcount batch process");
    
    
        }
    
        /**
         * 作用:将输入进来的每一行元素根据空格切分，切分成一个一个的单词
         * 再把切分好的单词变成《单词，1)这样的组合，元组中
         */
        static class SplitFun implements FlatMapFunction<String, Tuple2<String ,Integer>>{
    
            @Override
            public void flatMap(String in, Collector<Tuple2<String, Integer>> out) throws Exception {
                // a.将文本内容打散成一个一个单词
                String[] words = in.split(" "); //hello you ==> (hello,1) (you,1)
                for (String word : words) {
                    Tuple2<String, Integer> wordAndOne = new Tuple2<>();
                    out.collect(wordAndOne);
                }
            }
        }
    }

### 单词统计案例（流数据） ###

**需求**

统计一个文件中各个单词出现的次数，把统计结果输出到文件

**依赖**

**发送**

netcat下载：https://blog.csdn.net/qq\_38762390/article/details/115789281[Linux下载安装Netcat\_linux netcat下载\_人畜无害的萌新史莱姆的博客-CSDN博客![icon-default.png_t_N2N8][]https://blog.csdn.net/qq\_38762390/article/details/115789281][Linux_Netcat_linux netcat_-CSDN_icon-default.png_t_N2N8_https_blog.csdn.net_qq_38762390_article_details_115789281]

nc -lk 7777

**代码**

import org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    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;
    
    
    public class WordCountStream {
    
        public static void main(String[] args) {
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            // 获取流式数据
            String hostname = "192.168.25.129";
            int port = 7777;
            DataStreamSource<String> dSource = env.socketTextStream(hostname, port);
    
            SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = dSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                @Override
                public void flatMap(String in, Collector<Tuple2<String, Integer>> out) throws Exception {
                    String[] words = in.split(" ");
                    for (String word : words) {
                        Tuple2<String, Integer> wordAndOne = new Tuple2<>();
                        out.collect(wordAndOne);
                    }
                }
    
            });
    
            SingleOutputStreamOperator<Tuple2<String, Integer>> result = wordAndOne.keyBy(0).sum(1);
            result.print();
        }
    }

[icon-default.png_t_N2N8]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/611b81e4fd4345bfbb5528ba1bb59ee5.png
[Linux_Netcat_linux netcat_-CSDN_icon-default.png_t_N2N8_https_blog.csdn.net_qq_38762390_article_details_115789281]: https://blog.csdn.net/qq_38762390/article/details/115789281