Flink-05-DataStream API 之 Transformations

旧城等待， 2022-02-25 07:46 270阅读 0赞

**Flink拥有很多算子最常用的有如下：**

<table> 
 <thead> 
  <tr> 
   <th>算子</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>map：</td> 
   <td>输入一个元素，然后返回一个元素，中间可以做一些清洗转换等操作</td> 
  </tr> 
  <tr> 
   <td>flatmap：</td> 
   <td>输入一个元素，可以返回零个，一个或者多个元素</td> 
  </tr> 
  <tr> 
   <td>filter：</td> 
   <td>过滤函数，对传入的数据进行判断，符合条件的数据会被留下</td> 
  </tr> 
  <tr> 
   <td>keyBy：</td> 
   <td>根据指定的key进行分组，相同key的数据会进入同一个分区</td> 
  </tr> 
  <tr> 
   <td>reduce：</td> 
   <td>对数据进行聚合操作，结合当前元素和上一次reduce返回的值进行聚合操作，然后返回一个新的值</td> 
  </tr> 
  <tr> 
   <td>aggregations：</td> 
   <td>sum(),min(),max()等</td> 
  </tr> 
  <tr> 
   <td>Union：</td> 
   <td>合并多个流，新的流会包含所有流中的数据，但是union是一个限制，就是所有合并的流类型必须是一致的。</td> 
  </tr> 
  <tr> 
   <td>Connect：</td> 
   <td>和union类似，但是只能连接两个流，两个流的数据类型可以不同，会对两个流中的数据应用不同的处理方法。</td> 
  </tr> 
  <tr> 
   <td>CoMap,CoFlatMap：</td> 
   <td>在ConnectedStreams中需要使用这种函数，类似于map和flatmap</td> 
  </tr> 
  <tr> 
   <td>Split：</td> 
   <td>根据规则把一个数据流切分为多个流</td> 
  </tr> 
  <tr> 
   <td>Select：</td> 
   <td>和split配合使用，选择切分后的流</td> 
  </tr> 
 </tbody> 
</table>

##### 1、map： #####

输入一个元素，然后返回一个元素，中间可以做一些清洗转换等操作。  
输入流类型：DataStream  
输出流类型：DataStream  
例：对流入的数+1

DataStream<Integer> dataStream = //...
    dataStream.map(new MapFunction<Integer, Integer>() { 
        @Override
        public Integer map(Integer value) throws Exception { 
            return value + 1;
        }
    });

##### 2、flatmap： #####

输入一个元素，可以返回零个，一个或者多个元素。  
输入流类型：DataStream  
输出流类型：DataStream  
例：将输入的句子按空格分成多个单词。

DataStreamSource<String> text = env.socketTextStream(hostname,port,delimiter);
    DataStream  wordCountDataStream = text.flatMap(new FlatMapFunction<String, String>() { 
        @Override
        public void flatMap(String value, Collector<String> out)throws Exception { 
            for(String word: value.split(" ")){ 
                out.collect(word);
            }
        }
    });

##### 3、filter： #####

过滤函数，对传入的数据进行判断，符合条件的数据会被留下。  
输入流类型：DataStream  
输出流类型：DataStream  
例：过滤奇数，只显示偶数。

DataStream<Integer> dataStream = //...
    // 输入一串连续的数字 1 2 3 4 5 6 7 
    DataStream num = dataStream.map(new MapFunction<Integer, Integer>() { 
        @Override
        public Integer map(Integer value) throws Exception { 
            return value;
        }
    });
    // 执行 Filter 过滤 满足条件的数据会被留下
    DataStream result = num.filter(new FilterFunction<Integer>() { 
    	// 把所有的奇数过滤掉
        @Override
        public boolean filter(Integer value) throws Exception { 
            return value % 2 == 0;
        }
    });
    // 最后打印result的结果 是 2 4 6

##### 4、keyBy： #####

根据指定的key进行分组，相同key的数据会进入同一个分区。  
输入流类型：DataStream  
输出流类型：KeyedStream  
调用方式 dataStream.keyBy(“Word”)；  
参数说明：  
      参数只能是字段名称或者Tuple2的第几个元素，不可以用基本类型。  
      dataStream.keyBy(“Word”) // Key by field “Word”  
      dataStream.keyBy(0) // Key by the first element of a Tuple2  
注意:以下类型是无法作为key的

*  1、一个实体类对象，没有重写hashCode方法，并且依赖Object的hashCode方法
 *  2、一个任意形式的数组类型

例1：参数类型是类的一个属性（field ）  
定义一个类WordCount.java，有属性String word，和long count。来了一串英文句子，做单词计数。。

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);

例2：参数类型是Tuple2的第几个元素  
来了一串英文句子，做单词计数。。

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<Tuple2<String, Integer>> out) throws Exception { 
                    String[] splits = value.split("\\s");
                    for (String word:splits) { 
                        out.collect(new Tuple2(word,1));// 不一样的地方
                    }
                }
            }).keyBy(0)//Tuple2的第几个元素 **********重点***********
                .timeWindow(Time.seconds(2), Time.seconds(1))//指定计算数据的窗口大小（窗口的范围）和滑动窗口大小(多久滑动一次)
                .sum("count");//这里使用sum和reduce都可以
            // 把数据打印到控制台并且设置并行度
            wordCountDataStream.print().setParallelism(1);

##### 5、reduce： #####

对数据进行聚合操作，结合**当前元素和上一次reduce返回的值**进行聚合操作，然后返回一个新的值。\*\*sum()更方便。\*\*代码中有sum（）实现的方法，很简洁。  
输入流类型：KeyedStream  
输出流类型：DataStream  
例：对数据进行聚合操作  
例1：纯数字

keyedStream.reduce(new ReduceFunction<Integer>() { 
        @Override
        public Integer reduce(Integer value1, Integer value2)
        throws Exception { 
            return value1 + value2;
        }
    });

例2：单词计数  
同样：定义一个类WordCount.java，有属性String word，和long count。来了一串英文句子，做单词计数。。

DataStreamSource<String> text = env.socketTextStream(hostname,port,delimiter);
     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))//指定计算数据的窗口大小（窗口的范围）和滑动窗口大小(多久滑动一次)
          .reduce(new ReduceFunction<Integer>() { 
          //.sum("count");//这里使用sum和reduce都可以
    	    @Override
    	    public WordCount reduce(WordCount a,WordCount b) throws Exception { 
    	        return new WordCount(a.word,a.count+b.count);
    	    }
    	});
     
     // 把数据打印到控制台并且设置并行度
            wordCountDataStream.print().setParallelism(1);

##### 6、aggregations：聚合 #####

sum(),min(),max()等  
输入流类型：KeyedStream  
输出流类型：DataStream

keyedStream.sum(0);
    keyedStream.sum("key");
    keyedStream.min(0);
    keyedStream.min("key");
    keyedStream.max(0);
    keyedStream.max("key");
    keyedStream.minBy(0);
    keyedStream.minBy("key");
    keyedStream.maxBy(0);
    keyedStream.maxBy("key");

##### 7、Union： #####

合并多个流，新的流会包含所有流中的数据，但是union是一个限制，就是所有合并的**流类型**必须是**一致的。**  
场景：公司的数据来源可能是多个机器发送过来的，同一种数据存储在不同的机器，我们需要将这些数据分别读取，然后聚合在一起进行计算。

//获取Flink的运行环境
     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
     // 获取数据源 有两个数据源
     DataStreamSource<Long> text1 =  env.addSource(new MyNoParalleSource()).setParallelism(1);
     DataStreamSource<Long> text2 =  env.addSource(new MyNoParalleSource()).setParallelism(1);
     // 把text1和text2组装到一起
      DataStream<Long> text = text1.union(text2);

##### 8、Connect： #####

和union类似，但是只能连接两个流，两个流的数据类型可以不同，会对两个流中的数据应用不同的处理方法。  
输出流类型：ConnectedStreams<类型1 ,类型2 >

##### 9、CoMap,CoFlatMap： #####

跟Connect 配合使用，在ConnectedStreams中需要使用这种函数，类似于map和flatmap，我们知道Connect可以聚合两种数据类型，聚合成ConnectedStreams<类型1 ,类型2 >类型的数据流，当我们对这些数据进行map的时候就需要分别进行，所以在实现CoMap,CoFlatMap的时候，需要实现两个map（）方法；

// 获取数据源
            DataStreamSource<Long> text1 =  env.addSource(new MyNoParalleSource()).setParallelism(1);
            DataStreamSource<Long> text2 =  env.addSource(new MyNoParalleSource()).setParallelism(1);
            SingleOutputStreamOperator<String> text2_str =  text2.map(new MapFunction<Long, String>() { 
                @Override
                public String map(Long value) throws Exception { 
                    return "str_"  + value;
                }
            });
            ConnectedStreams<Long,String> connectStream =  text1.connect(text2_str);
            SingleOutputStreamOperator<Object> result = connectStream.map(new CoMapFunction<Long, String, Object>() { 
                @Override
                public Object map1(Long value) throws Exception { 
                // 处理类型1的数据 这里直接输出
                    return value;
                }
                @Override
                public Object map2(String value) throws Exception { 
                // 处理类型2的数据 这里直接输出
                    return value;
                }
            });
     
            //打印结果
             result.print().setParallelism(1);

##### 10、Split & Select #####

Split ：根据规则把一个数据流切分为多个流  
Select ：和split配合使用，选择切分后的流

// 对流进行切分，按照数据的奇偶性进行区分
            SplitStream<Long> splitStream = text.split(new OutputSelector<Long>() { 
                @Override
                public Iterable<String> select(Long value) { 
                    ArrayList<String> outPut = new ArrayList<>();
                    if(value %2 ==0){ 
                        outPut.add("even"); //偶数
                    }else{ 
                        outPut.add("odd"); //奇数
                    }
                    return outPut;
                }
            });
            // 选择一个或者多个切分后流
            DataStream<Long> evenStream = splitStream.select("even");
            DataStream<Long> oddStream = splitStream.select("odd");
            DataStream<Long> mixStream = splitStream.select("odd","even");
            //打印结果
            mixStream.print().setParallelism(1);