flink无状态转换算子map() flatMap()

浅浅的花香味﹌ 2022-12-01 01:21 191阅读 0赞

Apache Flink常见的一些场景为数据的ETL（抽取、转换、加载）管道任务。  
从一个或多个数据源获取数据，进行一些转换操作和信息补充，将结果存储起来。

## 无状态转换 ##

无状态的转换：包括map()和flatmap()

*  **map()**

调用用户定义的MapFunction对DataStream\[T\]数据进行处理，形成新的Data-Stream\[T\]，其中数据格式可能会发生变化，常用作对数据集内数据的清洗和转换。例如将输入数据集中的每个数值全部加1处理，并且将数据输出到下游数据集。  
map()基本是一对一服务，即输入一个元素输出一个元素。

例如：  
scala版本

val dataStream = env.fromElements(("a",1),("b",2),("a",3))
    val mapStream: DataStream[(String, Int)] = dataStream.map(t => (t_1, t_2 + 1)

使用map方法传入计算表达式，对第二个字段进行加1操作，然后得到新的mapStream

同时，也支持自定义实现MapFunction接口，编写自己的逻辑，做一些复杂的操作。  
例如：

// 通过指定MapFunction
    val mapStream: DataStream[(String, Int)] = dataStream.map(new MapFunction[(String, Int), (String, Int)] { 
      override def map(t: (String, Int)): (String, Int) = { 
        (t._1, t._2 + 1)}
      })

不过自定义实现MapFunction一般用于java。

DataStream<Integer> dataStream = env.fromElements(1, 2, 3, 4);
    DataStream<Integer> result = dataStream.map(new AddOne());

// 自定义MapFunction
    public static class AddOne implements MapFunction<Integer, Integer> {
        @Override
        public Integer map(Integer integer) throws Exception {
            return integer + 1;
        }
    }

输出为：

> 1> 5  
> 2> 2  
> 4> 4  
> 3> 3

*  **flatMap()**

对于map()来说，实现MapFunction也只是支持一对一的转换。  
那么有时候你需要处理一个输入元素，但是要输出一个或者多个输出元素的时候，就可以用到flatMap()。

void flatMap(T value,
                 Collector<O> out)
          throws Exception
          
    The core method of the FlatMapFunction. Takes an element from the input data set and transforms it into zero, one, or more elements.
    Parameters:
    value - The input value.
    out - The collector for returning result values.

> FlatMapFunction<T,O>  
> T - Type of the input elements.  
> O - Type of the returned elements.  
> 其中T为输入类型，O为输出类型

例如，对每一行的文本进行切割（如hive中的一行数据通常用，进行分割）。

scala版本

val dataStream:DataStream[String] = environment.fromCollections()
    val resultStream[String] = dataStream.flatMap {  str => str.split(" ") }

java版本

public static void main(String[] args) throws Exception { 
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            DataStream<String> dataStream = env.fromElements("a,b", "c,d,e", "f,g");
    
            DataStream<String> dataStream1 = dataStream.flatMap(new SplitData());
            dataStream1.print();
            env.execute("s");
    
    
        }
    
        public static class SplitData implements FlatMapFunction<String, String> { 
            @Override
            public void flatMap(String s, Collector<String> collector) throws Exception { 
                for (String key : s.split(",")) { 
                    collector.collect(key);
                }
            }
        }

输出为：

> 3> a  
> 3> b  
> 1> f  
> 1> g  
> 4> c  
> 4> d  
> 4> e

使用接口中提供的 Collector ，flatmap() 可以输出你想要的任意数量的元素，也可以一个都不发。

*另外还有个Filter*

*  **Filter**

DataStream->DataStream

根据条件进行判断，用于满足条件的数据进行输出，不满足就过滤掉。

java版本

DataStream<Integer> dataStream2 = env.fromElements(1, 2, 3, 4);
    
            DataStream<Integer> dataStream3 = dataStream2.filter((FilterFunction<Integer>) (num) -> { 
                return num - 2 > 0;
            });

scala版本

val filter: DataStream[Int] = dataStream.filter(_ - 2 > 0)

输出为：

> 1> 3  
> 2> 4