Flink操练（十八）之流处理单词统计

　　1、pom.xml文件<?xml version=＂1.0＂ encoding=＂UTF-8＂?>      4.0.0      org.example     Flink_learn     1.0-SNAPSHOT              1.13.0         1.8         2.12         1.7.30                                 org.apache.flink             flink-java             ${flink.version}                               org.apache.flink             flink-streaming-java_${scala.binary.version}             ${flink.version}                               org.apache.flink             flink-clients_${scala.binary.version}             ${flink.version}                               org.apache.flink             flink-table-api-java-bridge_${scala.binary.version}             ${flink.version}                               org.apache.flink             flink-table-planner-blink_${scala.binary.version}             ${flink.version}                               org.apache.flink             flink-streaming-scala_${scala.binary.version}             ${flink.version}                               org.apache.flink             flink-table-common             ${flink.version}                               org.apache.flink             flink-cep_${scala.binary.version}             ${flink.version}                               org.apache.flink             flink-csv             ${flink.version}                               org.apache.flink             flink-connector-kafka_${scala.binary.version}             ${flink.version}                               org.apache.bahir             flink-connector-redis_2.11             1.0                               mysql             mysql-connector-java             8.0.21                               org.apache.flink             flink-connector-jdbc_${scala.binary.version}             ${flink.version}                               org.slf4j             slf4j-log4j12             ${slf4j.version}                               org.apache.logging.log4j             log4j-to-slf4j             2.14.0                                                           org.apache.maven.plugins                 maven-assembly-plugin                 3.3.0                                                               jar-with-dependencies                                                                                                     make-assembly                         package                                                      single                                                                                                          org.apache.maven.plugins                 maven-compiler-plugin                                      8                     8                                              2、代码实现package one; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.common.functions.ReduceFunction; import org.apache.flink.api.java.functions.KeySelector; import org.apache.flink.streaming.api.datastream.DataStream; 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;  /**  * @program: Flink_learn  * @description: 从socket读取数据然后处理  * @author: Mr.逗  * @create: 2021-09-14 15:20  **/ public class Example1 {     public static void main(String[] args) {         // 获取流处理的运行时环境         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();         //设置并行度为1         env.setParallelism(1);         //读取数据源         //先在终端启动 `nc -lk 9999`         DataStreamSource source = env.socketTextStream(＂172.17.0.50＂, 9999);         // map操作         // 这里使用的flatMap方法         // map: 针对流中的每一个元素，输出一个元素         // flatMap：针对流中的每一个元素，输出0个，1个或者多个元素         DataStream mappedStream = source                 // 输入泛型：String; 输出泛型：WordWithCount                 .flatMap(new FlatMapFunction() {                     @Override                     public void flatMap(String v, Collector out) throws Exception {                         String[] words = v.split(＂,＂);                         //使用collect方法向下游发送数据                         for (String word : words) {                             out.collect(new WordWithCount(word, 1L));                         }                     }                 });         //分组shuffle         KeyedStream keyedStream = mappedStream.keyBy(new KeySelector() {             @Override             public String getKey(WordWithCount v) throws Exception {                 return v.word;             }         });         // reduce操作         // reduce会维护一个累加器         // 第一条数据到来，作为累加器输出         // 第二条数据到来，和累加器进行聚合操作，然后输出累加器         // 累加器和流中元素的类型是一样的         SingleOutputStreamOperator reduce = keyedStream.reduce(new ReduceFunction() {             @Override             public WordWithCount reduce(WordWithCount v1, WordWithCount v2) throws Exception {                 return new WordWithCount(v1.word, v1.count + v2.count);             }         });         //输出         reduce.print();         String name = Example1.class.getName();         //执行程序         try {             env.execute(name);         }catch (Exception e)         {             e.printStackTrace();         }      }     // POJO类     // 1. 必须是公有类     // 2. 所有字段必须是public     // 3. 必须有空构造器     // 模拟了case class     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 +                     ＂}＂;         }     } } 3、终端输入nc -lk 9999[xgsUser@fraud50 ~]$ nc -lk 9999 a,b a,c a,d a,a4、结果展示WordWithCount{word=＂�a＂, count=1} WordWithCount{word=＂b＂, count=1} WordWithCount{word=＂a＂, count=1} WordWithCount{word=＂c＂, count=1} WordWithCount{word=＂a＂, count=2} WordWithCount{word=＂d＂, count=1} WordWithCount{word=＂a＂, count=3} WordWithCount{word=＂a＂, count=4}