4.1.11 Flink-流处理框架-Flink流处理API之Transform转换算子

我就是我 2022-10-31 04:13 245阅读 0赞

**目录**

1.基本转换算子

1.1 map

1.2 flatMap扁平化算子

1.3 Filter过滤算子

1.4 基本算子代码演示

2.聚合算子

2.1 KeyBy

2.2 滚动聚合算子（Rolling Aggregation）

2.3 Reduce聚合

3.多流转换算子

3.1 Split 和 Select

3.1.1 Split分割

3.1.2 Select选择

3.2 Connect 和 CoMap

3.2.1 Connect

3.2.2 CoMap,CoFlatMap

3.3 Union

3.4 split+select+connect+union演示样例

需求：传感器数据按照温度高低（以 30 度为界），拆分成两个流。

--------------------

# 1.基本转换算子 #

## 1.1 map ##

map算子属于基本转换算子，来一个数据处理一个。

![20210221163135548.png][]

DataStream<Integer> mapStram = dataStream.map(new MapFunction<String, Integer>() {
     public Integer map(String value) throws Exception {
     return value.length();
     }
    });

## 1.2 flatMap扁平化算子 ##

DataStream<String> flatMapStream = dataStream.flatMap(new FlatMapFunction<String,
    String>() {
     public void flatMap(String value, Collector<String> out) throws Exception {
     String[] fields = value.split(",");
     for( String field: fields )
     out.collect(field);
     }
    });

## 1.3 Filter过滤算子 ##

![20210221163236641.png][]

DataStream<Interger> filterStream = dataStream.filter(new FilterFunction<String>()
    {
     public boolean filter(String value) throws Exception {
     return value == 1;
     }
    });

## 1.4 基本算子代码演示 ##

package com.atguigu.apitest.transform;
    import org.apache.flink.api.common.functions.FilterFunction;
    import org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.util.Collector;
    
    public class TransformTest1_Base {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // 从文件读取数据
            DataStream<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
    
            // 1. map，把String转换成长度输出
            DataStream<Integer> mapStream = inputStream.map(new MapFunction<String, Integer>() {
                @Override
                public Integer map(String value) throws Exception {
                    return value.length();
                }
            });
    
            // 2. flatmap，按逗号分字段
            DataStream<String> flatMapStream = inputStream.flatMap(new FlatMapFunction<String, String>() {
                @Override
                public void flatMap(String value, Collector<String> out) throws Exception {
                    String[] fields = value.split(",");
                    for( String field: fields )
                        out.collect(field);
                }
            });
    
            // 3. filter, 筛选sensor_1开头的id对应的数据
            DataStream<String> filterStream = inputStream.filter(new FilterFunction<String>() {
                @Override
                public boolean filter(String value) throws Exception {
                    return value.startsWith("sensor_1");
                }
            });
    
            // 打印输出
            mapStream.print("map");
            flatMapStream.print("flatMap");
            filterStream.print("filter");
    
            env.execute();
        }
    }

# 2.聚合算子 #

dataStream没有办法直接做聚合操作，需要借助于keyedStream分组之后再进行聚合操作。

## 2.1 KeyBy ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

DataStream → KeyedStream：**逻辑地将一个流拆分成不相交的分区，每个分区包含具有相同 key 的元素，在内部以 hash 的形式实现的**。具有相同key的数据肯定都在一个分区，同一个分区下面不一定只含有一个key。

## 2.2 滚动聚合算子（Rolling Aggregation） ##

这些算子可以针对 KeyedStream 的每一个支流做聚合。

⚫ sum() ⚫ min() ⚫ max()⚫ minBy() ⚫ maxBy()

package com.atguigu.apitest.transform;
    import com.atguigu.apitest.beans.SensorReading;
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.java.functions.KeySelector;
    import org.apache.flink.api.java.tuple.Tuple;
    import org.apache.flink.streaming.api.datastream.DataStream;
    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 sun.awt.SunHints;
    import javax.xml.crypto.Data;
    
    public class TransformTest2_RollingAggregation {
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(4);
    
            // 从文件读取数据
            DataStream<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
    
            // 转换成SensorReading类型
    //        DataStream<SensorReading> dataStream = inputStream.map(new MapFunction<String, SensorReading>() {
    //            @Override
    //            public SensorReading map(String value) throws Exception {
    //                String[] fields = value.split(",");
    //                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
    //            }
    //        });
    
            DataStream<SensorReading> dataStream = inputStream.map( line -> {
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            } );
    
            // 分组
            KeyedStream<SensorReading, Tuple> keyedStream = dataStream.keyBy("id");
            KeyedStream<SensorReading, String> keyedStream1 = dataStream.keyBy(data -> data.getId());
    
            DataStream<Long> dataStream1 = env.fromElements(1L, 34L, 4L, 657L, 23L);
            KeyedStream<Long, Integer> keyedStream2 = dataStream1.keyBy(new KeySelector<Long, Integer>() {
                @Override
                public Integer getKey(Long value) throws Exception {
                    return value.intValue() % 2;
                }
            });
    
            //        KeyedStream<SensorReading, String> keyedStream1 = dataStream.keyBy(SensorReading::getId);
    
            // 滚动聚合，取当前最大的温度值
            DataStream<SensorReading> resultStream = keyedStream.maxBy("temperature");
    
            resultStream.print("result");
    
            keyedStream1.print("key1");
            keyedStream2.sum(0).print("key2");
            env.execute();
        }
    }

max函数和maxBy函数都是滚动聚合求最大值，当输入一条新的数据的时候，会重新计算最大值，如果有改动会重新输出最大值。max函数的缺点是只是更新了温度值，但是其他属性比如时间戳是不变的，这个显然在业务逻辑中是有问题的，我们可以使用maxBy避免这个问题，maxBy可以更新温度值和时间戳。

## 2.3 Reduce聚合 ##

KeyedStream → DataStream：一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果。比如下面代码：取最大的温度值，以及当前最新的时间戳。

package com.atguigu.apitest.transform;
    
    import com.atguigu.apitest.beans.SensorReading;
    import org.apache.flink.api.common.functions.ReduceFunction;
    import org.apache.flink.api.java.tuple.Tuple;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.KeyedStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    public class TransformTest3_Reduce {
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // 从文件读取数据
            DataStream<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
    
            // 转换成SensorReading类型
            DataStream<SensorReading> dataStream = inputStream.map(line -> {
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            // 分组
            KeyedStream<SensorReading, Tuple> keyedStream = dataStream.keyBy("id");
    
            // reduce聚合，取最大的温度值，以及当前最新的时间戳
            SingleOutputStreamOperator<SensorReading> resultStream = keyedStream.reduce(new ReduceFunction<SensorReading>() {
                @Override
                public SensorReading reduce(SensorReading value1, SensorReading value2) throws Exception {
                    return new SensorReading(value1.getId(), value2.getTimestamp(), Math.max(value1.getTemperature(), value2.getTemperature()));
                }
            });
    
            keyedStream.reduce( (curState, newData) -> {
                return new SensorReading(curState.getId(), newData.getTimestamp(), Math.max(curState.getTemperature(), newData.getTemperature()));
            });
    
            resultStream.print();
            env.execute();
        }
    }

# 3.多流转换算子 #

## 3.1 Split 和 Select ##

### 3.1.1 Split分割 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1][]

DataStream → SplitStream：根据某些特征把一个 DataStream 拆分成两个或者 多个 DataStream。

### 3.1.2 Select选择 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2][]

SplitStream→DataStream：从一个 SplitStream 中获取一个或者多个 DataStream。

## 3.2 Connect 和 CoMap ##

### 3.2.1 Connect ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 3][]

DataStream,DataStream → ConnectedStreams：连接两个保持他们类型的数 据流，两个数据流被 Connect 之后，只是被放在了一个同一个流中，内部依然保持 各自的数据和形式不发生任何变化，两个流相互独立。

### 3.2.2 CoMap,CoFlatMap ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 4][]

ConnectedStreams → DataStream：作用于 ConnectedStreams 上，功能与 map 和 flatMap 一样，对 ConnectedStreams 中的每一个 Stream 分别进行 map 和 flatMap 处理。

## 3.3 Union ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 5][]

DataStream → DataStream：对两个或者两个以上的 DataStream 进行 union 操 作，产生一个包含所有 DataStream 元素的新 DataStream。

> Connect 与 Union 区别：
> 
>  *  1.Union 之前两个流的类型必须是一样，Connect 可以不一样，在之后的 coMap 中再去调整成为一样的。
>  *  2.Connect 只能操作两个流，Union 可以操作多个。

## 3.4 split+select+connect+union演示样例 ##

### 需求：传感器数据按照温度高低（以 30 度为界），拆分成两个流。 ###

package com.atguigu.apitest.transform;
    
    import com.atguigu.apitest.beans.SensorReading;
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.api.java.tuple.Tuple3;
    import org.apache.flink.streaming.api.collector.selector.OutputSelector;
    import org.apache.flink.streaming.api.datastream.ConnectedStreams;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.datastream.SplitStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.co.CoMapFunction;
    
    import java.util.Collections;
    
    
    public class TransformTest4_MultipleStreams {
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // 从文件读取数据
            DataStream<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
    
            // 转换成SensorReading
            DataStream<SensorReading> dataStream = inputStream.map(line -> {
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            } );
    
            // 1. 分流，按照温度值30度为界分为两条流
            SplitStream<SensorReading> splitStream = dataStream.split(new OutputSelector<SensorReading>() {
                @Override
                public Iterable<String> select(SensorReading value) {
                    return (value.getTemperature() > 30) ? Collections.singletonList("high") : Collections.singletonList("low");
                }
            });
    
            DataStream<SensorReading> highTempStream = splitStream.select("high");
            DataStream<SensorReading> lowTempStream = splitStream.select("low");
            DataStream<SensorReading> allTempStream = splitStream.select("high", "low");
    
            highTempStream.print("high");
            lowTempStream.print("low");
            allTempStream.print("all");
    
            // 2. 合流 connect，将高温流转换成二元组类型，与低温流连接合并之后，输出状态信息
            DataStream<Tuple2<String, Double>> warningStream = highTempStream.map(new MapFunction<SensorReading, Tuple2<String, Double>>() {
                @Override
                public Tuple2<String, Double> map(SensorReading value) throws Exception {
                    return new Tuple2<>(value.getId(), value.getTemperature());
                }
            });
    
            ConnectedStreams<Tuple2<String, Double>, SensorReading> connectedStreams = warningStream.connect(lowTempStream);
    
            DataStream<Object> resultStream = connectedStreams.map(new CoMapFunction<Tuple2<String, Double>, SensorReading, Object>() {
                @Override
                public Object map1(Tuple2<String, Double> value) throws Exception {
                    return new Tuple3<>(value.f0, value.f1, "high temp warning");
                }
    
                @Override
                public Object map2(SensorReading value) throws Exception {
                    return new Tuple2<>(value.getId(), "normal");
                }
            });
    
            resultStream.print();
    
            // 3. union联合多条流
    //        warningStream.union(lowTempStream);
            highTempStream.union(lowTempStream, allTempStream);
    
            env.execute();
        }
    }

[20210221163135548.png]: /images/20221024/06f18f3d71794684823ca38a3167e6f5.png
[20210221163236641.png]: /images/20221024/7f3cea6978434088a36f6381baba88f1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221024/61325b03aa85476ea2b113bed624a8e9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1]: /images/20221024/bc28b3ca357348d8b462d041e4c4340d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2]: /images/20221024/d9e24aedd4a14e01b81c7a181980268d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 3]: /images/20221024/74101634746344a99a10dc80be535640.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 4]: /images/20221024/d3fa61460cb5446e95337ae40c2d8c74.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 5]: /images/20221024/c6989d1f90c445ddbeb9a9e2052dfe0a.png