flink入门实例-Windows下本地模式跑SocketWordCount

蔚落 2023-06-19 11:26 349阅读 0赞

### 文章目录 ###

*   *  一、nc工具
     *  二、idea中配置，代码以及设置参数
     *  三、运行结果

## 一、nc工具 ##

配置windows的nc端口，在网上下载nc.exe(https://eternallybored.org/misc/netcat/)  
进入解压目录运行如下命令

nc -L -p 9000 -v

使用命令开始nc制定端口为9000（nc -L -p 9000 -v） 启动插件

## 二、idea中配置，代码以及设置参数 ##

maven配置：

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>lims</groupId>
        <artifactId>flink-project</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <properties>
            <flink.version>1.7.2</flink.version>
        </properties>
    
        <dependencies>
            
            <dependency>
                <groupId>log4j</groupId>
                <artifactId>log4j</artifactId>
                <version>1.2.17</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-java</artifactId>
                <version>${flink.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java_2.11</artifactId>
                <version>${flink.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-clients_2.11</artifactId>
                <version>${flink.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-connector-wikiedits_2.11</artifactId>
                <version>${flink.version}</version>
            </dependency>
        </dependencies>
    
    </project>

WordCount的源码：我这里直接把端口写死了

import org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.api.java.utils.ParameterTool;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.util.Collector;
    
    
    /**
     * @Description: TODO
     * @Date: 2019/12/5 20:49
     */
    public class WordCount {
        public static void main(String[] args) throws Exception {
           //定义socket的端口号
            int port = 9000;
            /*try{
                ParameterTool parameterTool = ParameterTool.fromArgs(args);
                port = parameterTool.getInt("port");
            }catch (Exception e){
                System.err.println("没有指定port参数，使用默认值9000");
                port = 9000;
            }*/
    
            //获取运行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            //连接socket获取输入的数据
            DataStreamSource<String> text = env.socketTextStream("127.0.0.1", port, "\n");
    
            //计算数据
            DataStream<WordWithCount> windowCount = text.flatMap(new FlatMapFunction<String, WordWithCount>() {
                public void flatMap(String value, Collector<WordWithCount> out) throws Exception {
                    String[] splits = value.split("\\s");
                    for (String word:splits) {
                        out.collect(new WordWithCount(word,1L));
                    }
                }
            })//打平操作，把每行的单词转为<word,count>类型的数据
                    .keyBy("word")//针对相同的word数据进行分组
                    .timeWindow(Time.seconds(2),Time.seconds(1))//指定计算数据的窗口大小和滑动窗口大小
                    .sum("count");
    
            //把数据打印到控制台
            windowCount.print()
                    .setParallelism(1);//使用一个并行度
            //注意：因为flink是懒加载的，所以必须调用execute方法，上面的代码才会执行
            env.execute("streaming word count");
    
        }
    
        /**
         * 主要为了存储单词以及单词出现的次数
         */
        public static class WordWithCount{
            public String word;
            public long count;
            public WordWithCount(){}
            public WordWithCount(String word, long count) {
                this.word = word;
                this.count = count;
            }
    
            @Override
            public String toString() {
                return "WordWithCount{" +
                        "word='" + word + '\'' +
                        ", count=" + count +
                        '}';
            }
        }
    }

## 三、运行结果 ##

cmd中输入单词，空格分割，并换行，在idea的控制台中观察输出  
![在这里插入图片描述][20191206150213878.png]  
IDEA![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70]  
本地开发调试实例完成

原文地址  
https://www.cnblogs.com/limaosheng/p/10434848.html

[20191206150213878.png]: https://img-blog.csdnimg.cn/20191206150213878.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191206150228790.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE=,size_16,color_FFFFFF,t_70