Flink-03-【入门程序】WordCount的流处理和批处理

ゞ浴缸里的玫瑰 2022-02-25 06:48 272阅读 0赞

### 需求分析 ###

手工通过socket试试产生一些单词，使用flink实时接收数据，对指定时间窗口内（例如：2秒）的数据进行聚合统计，并且把时间窗口内计算的结果打印出来。

#### 实现步骤 ####

1.  获得一个执行环境
2.  加载/创建初始化数据–连接socket获取输入的数据；
3.  指定操作数据的transaction算子
4.  指定把计算好的数据放在哪里
5.  调用execute（）触发执行程序

**flink程序是延迟计算的，只有最后调用execute（）方法的时候才会真正触发执行程序。  
延迟计算的好处：你可以开发复杂的程序，但是flink可以将复杂的程序转成一个plan，将plan作为一个整体单元执行。**

#### pom.xml ####

我使用java实现，如果使用Scala，导入相应包即可…

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

### 方式一：流处理 ###

package com.test;
     
    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;
     
    /** * @Author:renxin.tang * @Desc: * @Date: Created in 10:34 2019/3/28 */
    public class WindowWordCount { 
        //定义socket端口号
        public static void main(String[] args) throws Exception { 
            int port;
            try { 
                ParameterTool parameterTool = ParameterTool.fromArgs(args);
                port = parameterTool.getInt("port");
            }catch (Exception e){ 
                System.err.println("没有指定参数，使用默认值9001");
                port=9001;
            }
        //1、获取流处理运行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            String hostname = "hwyuntrx";//指定主机名/ip
            String delimiter = "\n";//指定行的结束符
        //2、加载/创建初始化数据--连接socket获取输入的数据
            DataStreamSource<String> text = env.socketTextStream(hostname,port,delimiter);
        //3、指定操作数据的transaction算子
            // 数据打平操作-- 流处理的返回值类型是DataStream 流处理是 DataSet
            DataStream<WordCount>wordCountDataStream = text.flatMap(new FlatMapFunction<String, WordCount>() { 
                @Override
                public void flatMap(String value, Collector<WordCount> out) throws Exception { 
                    String[] splits = value.split("\\s");
                    for (String word:splits) { 
                        out.collect(new WordCount(word,1));
                    }
                }
            }).keyBy("word")//根据那个属性进行分组--WordCount 中的word属性
                .timeWindow(Time.seconds(2), Time.seconds(1))//指定计算数据的窗口大小（窗口的范围）和滑动窗口大小(多久滑动一次)
                .sum("count");//这里使用sum和reduce都可以
            // 把数据打印到 控制台 并且设置并行度
            wordCountDataStream.print().setParallelism(1);
            //执行--
            env.execute("Socket word Count");
        }
        public static class WordCount{ 
            public String word;
            public long count;
            public WordCount(){ 
            }
            public WordCount(String word,long count){ 
                this.count = count;
                this.word = word;
            }
            @Override
            public String toString() { 
                return "WordCount{" +
                        "word='" + word + '\'' +
                        ", count=" + count +
                        '}';
            }
        }
     
    }

###### 测试： ######

1、开启主机名为hwyuntrx的云服务器（虚拟机也行）,端口可以自己制定，代码中的端口跟这个端口对应即可，输入如下命令：

nc -l 9001

2、idea启动flink程序  
3、在linux终端输入：  
![在这里插入图片描述][image2019-3-29_15-8-51.png_version_1_modificationDate_1553843331000_api_v2]  
运行结果：  
![在这里插入图片描述][image2019-3-29_15-9-28.png_version_1_modificationDate_1553843368000_api_v2]

### 方式二：批处理 ###

场景：从本地获取一个文件，程序读取文件中的信息，对获取的数据进行批量处理。

package com.test;
     
    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.DataSource;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.util.Collector;
     
    /** * @Author:renxin.tang * @Desc: * @Date: Created in 15:52 2019/3/28 */
    public class BatchWordCountJava { 
     
        public static void main(String[] args) throws Exception { 
            // 获取批处理运行环境
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
            String inputPath="D:\\DATA\\test.txt";// 指定输入文件路径
            String outPath = "D:\\DATA\\out.txt";
            DataSource<String> text =  env.readTextFile(inputPath);// 获取文件中的内容
     
            // 对数据内容进行处理 ---打平 批处理返回类型是DataSet 流处理是DataStream
            DataSet<Tuple2<String, Integer>> counts = text.flatMap(new Tokenizer()).groupBy(0).sum(1);
     
            // .setParallelism(1) 设置并行度为1 不然输出的结果会是一个文件夹 这个文件夹下是有多个输出的文件，文件的数量取决于你计算机cpu的核数
            counts.writeAsCsv(outPath,"\n"," ").setParallelism(1);
            env.execute("Batch Word Count");
     
        }
        public static class Tokenizer implements FlatMapFunction<String, Tuple2<String,Integer>> { 
     
     
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception { 
                String[] splits = value.toLowerCase().split("\\s+");// 转成小写切分 即不区分大小写
                for (String word:splits) { 
                    out.collect(new Tuple2(word,1));
                }
            }
        }
     
     
     
    }

**测试：**

在电脑上本地有D:\\DATA\\test.txt文件读入，输出的结果将放在D:\\DATA\\out.txt，如果没有指定并行度为1（.setParallelism(1)），输出的是一个文件夹，这个文件夹里面会有几个文件，分别存储处理完的结果，文件的数量取决于你电脑的CPU核数。

#### 总结 ####

1、程序实现的步骤：

1）、获得一个执行环境  
      2）、加载/创建初始化数据–连接socket获取输入的数据；  
      3）、指定操作数据的transaction算子  
      4）、指定把计算好的数据放在哪里  
      5）、调用execute（）触发执行程序

2、Flink Streaming和batch的区别：

*  流处理Streaming
    
        运行环境：StreamExecutionEnvironment
              数据源类型：DataStreamSource
              结果集类型：DataStream
 *  批处理Batch
    
        运行环境：ExecutionEnvironment
               数据源类型：DataSource
               结果集类型：DataSet

[image2019-3-29_15-8-51.png_version_1_modificationDate_1553843331000_api_v2]: http://wiki.corp.qunar.com/confluence/download/attachments/244739405/image2019-3-29_15-8-51.png?version=1&modificationDate=1553843331000&api=v2
[image2019-3-29_15-9-28.png_version_1_modificationDate_1553843368000_api_v2]: http://wiki.corp.qunar.com/confluence/download/attachments/244739405/image2019-3-29_15-9-28.png?version=1&modificationDate=1553843368000&api=v2