Flink操练（二十）之并行度使用讲解

　　1、代码实现逻辑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.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: 并行度的设置  *  针对每个算子设置的并行度的优先级高于全局并行度  * 本程序需要两个任务插槽  * @author: Mr.逗  * @create: 2021-09-14 15:40  **/ public class Example3 {     public static void main(String[] args) {         // 获取流处理的运行时环境         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();         // 设置并行任务的数量为1         // 需要1个任务插槽         env.setParallelism(1);         //读取数据源         // 并行度设置为1         DataStreamSource stream = env.fromElements(＂hello world＂, ＂hello world＂).setParallelism(1);         // map操作         // 这里使用的flatMap方法         // map: 针对流中的每一个元素，输出一个元素         // flatMap：针对流中的每一个元素，输出0个，1个或者多个元素         // 并行度设置为2         SingleOutputStreamOperator mappedStream = stream                 // 输入泛型：String; 输出泛型：WordWithCount                 .flatMap(new FlatMapFunction() {                     @Override                     public void flatMap(String v, Collector out) throws Exception {                         String[] words = v.split(＂ ＂);                         for (String w : words) {                             // 使用collect方法向下游发送数据                             out.collect(new WordWithCount(w, 1L));                         }                     }                 }).setParallelism(2);         //分组shuffle         // 第一个泛型：流中元素的泛型         // 第二个泛型：key的泛型         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 = Example3.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 +                     ＂}＂;         }     } } 2、结果展示WordWithCount{word=＂hello＂, count=1} WordWithCount{word=＂world＂, count=1} WordWithCount{word=＂hello＂, count=2} WordWithCount{word=＂world＂, count=2}