Flink：Streaming+Batch 完成WordCount

水深无声 2022-09-09 06:41 214阅读 0赞

#### Flink Streaming和Batch的区别 ####

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th>流处理Streaming</th> 
   <th>批处理Batch</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>执行环境</td> 
   <td>StreamExecutionEnvironment</td> 
   <td>ExecutionEnvironment</td> 
  </tr> 
  <tr> 
   <td>数据类型</td> 
   <td>DataStream</td> 
   <td>DataSet</td> 
  </tr> 
 </tbody> 
</table>

#### Streaming ####

我们通过socket实时产生一些单词，使用flink实时接收数据，对指定时间窗口内(例如:2秒)的数据进行聚合 统计，并且把时间窗口内计算的结果打印出来

**Scala版本**

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.streaming.api.windowing.time.Time
    
    
    
    object SocketWindowWordCountScala { 
      	def main(args: Array[String]): Unit = { 
    	    //获取运行环境
    	    val env = StreamExecutionEnvironment.getExecutionEnvironment
    	
    	    //连接socket获取输入数据
    	    val text = env.socketTextStream("bigdata", 9001)
    	
    	    //处理数据
    	    //注意：必须要添加这一行隐式转换的代码，否则下面的flatMap方法会报错
    	    import org.apache.flink.api.scala._
    	    val wordCount = text.flatMap(_.split(" "))	// 将每一行数据根据空格切分单词
    	      .map((_,1))	// 每一个单词转换为tuple2的形式(单词，1)
    	      .keyBy(tup=>tup._1)	// 官方推荐使用keySelector选择器选择数据
    	      .timeWindow(Time.seconds(2))	// 时间窗口为2秒，表示每隔2秒钟计算一次接收到的数据
    	      .sum(1)	// 使用sum或者reduce都可以
    	      //.reduce((t1,t2)=>(t1._1,t1._2+t2._2))
    	
    	    //使用一个线程执行打印操作
    	    wordCount.print().setParallelism(1)
    	
    	    //执行程序
    	    env.execute("SocketWindowWordCountScala")
    
      	}
    
    }

**Java版本**

import org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.api.common.functions.MapFunction;
    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.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.util.Collector;
    
    
    public class SocketWindowWordCountJava { 
        public static void main(String[] args) throws Exception{ 
            //获取运行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            //连接socket获取输入数据
            DataStreamSource<String> text = env.socketTextStream("bigdata0", 9001);
    
            //处理数据
            SingleOutputStreamOperator<Tuple2<String, Integer>> wordCount = text.flatMap(new FlatMapFunction<String, String>() { 
                public void flatMap(String line, Collector<String> out) throws Exception { 
                    String[] words = line.split(" ");
                    for (String word : words) { 
                        out.collect(word);
                    }
                }
            }).map(new MapFunction<String, Tuple2<String, Integer>>() { 
                public Tuple2<String, Integer> map(String word) throws Exception { 
                    return new Tuple2<String, Integer>(word, 1);
                }
            }).keyBy(new KeySelector<Tuple2<String, Integer>, String>() { 
                public String getKey(Tuple2<String, Integer> tup) throws Exception { 
                    return tup.f0;
                }
            })//.keyBy(0)
                    .timeWindow(Time.seconds(2))
                    .sum(1);
    
            //使用一个线程执行打印操作
            wordCount.print().setParallelism(1);
    
            //执行程序
            env.execute("SocketWindowWordCountJava");
    
        }
    }

我们通过命令行像9001端口发送数据  
![在这里插入图片描述][c8ab2267a37149c089cb34d90cd4e733.png]  
能看到控制台打印结果  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_18_color_FFFFFF_t_70_g_se_x_16]

#### Batch ####

**Scala版本**

import org.apache.flink.api.scala.ExecutionEnvironment
    
    
    object BatchWordCountScala { 
      	def main(args: Array[String]): Unit = { 
    	    //获取执行环境
    	    val env = ExecutionEnvironment.getExecutionEnvironment
    	
    	    val inputPath = "hdfs://bigdata:9000/hello.txt"
    	    val outPath = "hdfs://bigdata:9000/out"
    	
    	    //读取文件中的数据
    	    val text = env.readTextFile(inputPath)
    	
    	    //处理数据
    	    import org.apache.flink.api.scala._
    	    val wordCount = text.flatMap(_.split(" "))
    	      .map((_, 1))
    	      .groupBy(0)
    	      .sum(1)
    	      .setParallelism(1)//这里面设置并行度为1是为了将所有数据写到一个文件里面，查看结果比较方便
    	
    	    //将结果数据保存到文件中
    	    wordCount.writeAsCsv(outPath,"\n"," ")
    	
    	    //执行程序
    	    env.execute("BatchWordCountScala")
      	}
    }

**Java版本**

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.tuple.Tuple2;
    import org.apache.flink.util.Collector;
    
    /** * 需求：统计指定文件中单词出现的总次数 */
    public class BatchWordCountJava { 
        public static void main(String[] args) throws Exception{ 
            //获取执行环境
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    
            String inputPath = "hdfs://bigdata:9000/hello.txt";
            String outPath = "hdfs://bigdata:9000/out2";
    
            //读取文件中的数据
            DataSource<String> text = env.readTextFile(inputPath);
    
            //处理数据
            AggregateOperator<Tuple2<String, Integer>> wordCount = text.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() { 
                public void flatMap(String line, Collector<Tuple2<String, Integer>> out)
                        throws Exception { 
                    String[] words = line.split(" ");
                    for (String word : words) { 
                        out.collect(new Tuple2<String, Integer>(word, 1));
                    }
                }
            }).groupBy(0)
                    .sum(1)
                    .setParallelism(1);
    
            //将结果数据保存到文件中
            wordCount.writeAsCsv(outPath,"\n"," ");
    
            //执行程序
            env.execute("BatchWordCountJava");
        }
    }

[c8ab2267a37149c089cb34d90cd4e733.png]: /images/20220829/681950344a2344eab2f9751d84080651.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA566A5Y2V6ZqP6aOO_size_18_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/7021849e76cb42649af6d1eee7e2c877.png