Flink 流处理API

爱被打了一巴掌 2022-11-16 08:32 227阅读 0赞

## 一、Environment ##

### 1.1 getExecutionEnvironment ###

创建一个执行环境，表示当前执行程序的上下文。如果程序是独立调用的，则此方法返回本地执行环境；如果从命令行客户端调用程序以提交到集群，则此方法返回此集群的执行环境。也就是说，getExecutionEnvironment会根据查询运行的方式决定返回什么样的运行环境，是最常用的一种创建执行环境的方式。

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

如果没有设置并行度，会以 flink-conf.yaml 中的配置为准，默认是1

在这里插入代码片

### 1.2 createLocalEnvironment ###

返回本地执行环境，需要在调用时指定默认的并行度

LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1);

### 1.3 createRemoteEnvironment ###

返回集群执行环境，将Jar提交到远程服务器。需要在调用时指定 JobManager的 IP 和 端口号 ，并指定要在集群中运行的Jar包

StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("jobmanage-hostname", 6123, "Jar包路径");

## 二、Source ##

### 2.1 从集合读取数据 ###

public class SourceTest1_Collection { 
        public static void main(String[] args) throws Exception{ 
            // 创建执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            // 从集合中读取数据，DataStreamSource 可以使用 DataStream类 替换
            DataStreamSource<SensorReading> dataStreamSource = env.fromCollection(
                    Arrays.asList(
                            new SensorReading("sensor_1", 154777324345L, 35.8),
                            new SensorReading("sensor_2", 154777324346L, 15.4),
                            new SensorReading("sensor_3", 154777324347L, 6.7),
                            new SensorReading("sensor_4", 154777324348L, 38.1)
                    )
            );
    
            // 也可以使用如下方法
            DataStream<Integer> integerDataStream = env.fromElements(1,2,3,10,2324);
    
            // 打印输出
            dataStreamSource.print("data");
            integerDataStream.print("int");
    
            // 执行，参数是Jobname，有默认值
            env.execute();
        }
    }

打印结果：

int:4> 2
    int:1> 3
    data:4> com.test.sun.apitest.beans.SensorReading@c721304
    data:1> com.test.sun.apitest.beans.SensorReading@7aef3173
    int:3> 1
    data:2> com.test.sun.apitest.beans.SensorReading@1f657d1e
    data:3> com.test.sun.apitest.beans.SensorReading@347daeb1
    int:2> 10
    int:3> 2324

从打印结果可以看出来，集合数据是乱序的。若想保证打印结果是按集合顺序的，可以使用`env.setParallelism(1)`方法，完整代码如下所示

public class SourceTest1_Collection { 
        public static void main(String[] args) throws Exception{ 
            // 创建执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            // 如果想要输出结果是按顺序的，可以使用如下方式
            env.setParallelism(1);
    
            // 从集合中读取数据，DataStreamSource 可以使用 DataStream类 替换
            DataStreamSource<SensorReading> dataStreamSource = env.fromCollection(
                    Arrays.asList(
                            new SensorReading("sensor_1", 154777324345L, 35.8),
                            new SensorReading("sensor_2", 154777324346L, 15.4),
                            new SensorReading("sensor_3", 154777324347L, 6.7),
                            new SensorReading("sensor_4", 154777324348L, 38.1)
                    )
            );
    
            // 也可以使用如下方法
            DataStream<Integer> integerDataStream = env.fromElements(1,2,3,10,2324);
    
            // 打印输出
            dataStreamSource.print("data");
            integerDataStream.print("int");
    
            // 执行，参数是Jobname，有默认值
            env.execute();
        }
    }

打印结果：

int> 1
    int> 2
    int> 3
    data> com.test.sun.apitest.beans.SensorReading@4754ae8e
    int> 10
    data> com.test.sun.apitest.beans.SensorReading@49935814
    int> 2324
    data> com.test.sun.apitest.beans.SensorReading@4dacc5fe
    data> com.test.sun.apitest.beans.SensorReading@8ae522f

### 2.2 从文件读取数据 ###

public class SourceTest2_File { 
        public static void main(String[] args) throws Exception{ 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            // 从文件中读取数据
            DataStream<String> dataStream = env.readTextFile("/Users/sundongping/IdeaProjects/FirstFlinkTest/src/main/resources/sensor.txt");
    
            // 打印输出
            dataStream.print();
    
            env.execute();
        }
    }

打印结果：

4> sensor_1,154777324345L,35.8
    2> sensor_4,154777324348L,38.1
    1> sensor_3,154777324347L,6.7
    4> sensor_2,154777324346L,15.4

### 2.3 从Kafka消息队列读取数据 ###

#### 2.3.1 需要引入kafka连接器的依赖 ####

<dependency>
    	<groupId>org.apache.flink</groupId>
     	<artifactId>flink-connector-kafka-0.11_2.12</artifactId>
        <version>1.9.0</version>
    </dependency>

#### 2.3.2 SourceTest3\_Kafka类 ####

public class SourceTest3_Kafka { 
        public static void main(String[] args) throws Exception{ 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            // kafka配置项
            Properties properties = new Properties();
            properties.setProperty("bootstrap.servers", "localhost:9092");
            properties.setProperty("group.id", "consumer-group");
            properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            properties.setProperty("auto.offset.reset", "latest");
    
            // 从kafka中读取数据
            DataStream<String> dataStream = env.addSource(new FlinkKafkaConsumer011<String>("sensor", new SimpleStringSchema(), properties));
    
            // 打印输出
            dataStream.print();
    
            env.execute();
        }
    }

#### 2.3.3 启动zookeeper ####

cd zookeeper-3.4.13
    ./bin/zkServer.sh start

出现如下信息表示启动成功

ZooKeeper JMX enabled by default
    Using config: /Users/sundongping/Downloads/安装包/zookeeper-3.4.13/bin/../conf/zoo.cfg
    Starting zookeeper ... already running as process 26287.

第一次启动zookeeper的话，通常会失败，失败信息如下

ZooKeeper JMX enabled by default
    Using config: /Users/sundongping/Downloads/安装包/zookeeper-3.4.13/bin/../conf/zoo.cfg
    grep: /Users/sundongping/Downloads/安装包/zookeeper-3.4.13/bin/../conf/zoo.cfg: No such file or directory
    mkdir: : No such file or directory
    Starting zookeeper ... ./bin/zkServer.sh: line 149: /zookeeper_server.pid: Permission denied
    FAILED TO WRITE PID

解决这个报错的方法是在**conf目录下，新建一个zoo.cfg文件**，文件内容如下

tickTime=2000
    initLimit=10
    syncLimit=5
    dataDir=/Users/sundongping/Downloads/安装包/zookeeper-3.4.13
    clientPort=2181

#### 2.3.4 启动kafka ####

cd kafka_2.11-2.1.0
    ./bin/kafka-server-start.sh -daemon ./config/server.properties
    ./bin/./kafka-console-producer.sh --broker-list 192.168.1.5:9092 --topic sensor

⚠️注意：kafka与zookeeper的版本必须对应，否则kafka会报如下错误`could not be established. Broker may not be available. (org.apache.kafka.cli`。如果你不知道自己的kafka版本对应的zookeeper版本号的话，可以去`kafka_2.11-2.1.0/libs`目录下看kafka自带的zookeeper版本

## 2.4 自定义测试数据源 ##

自定义数据源需要做的只是传入一个 SourceFunction就可以。

public static void main(String[] args) throws Exception{ 
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
        // 从kafka中读取数据
        DataStream<SensorReading> dataStream = env.addSource(new MySensorSource());
    
        // 打印输出
        dataStream.print();
    
        env.execute();
    }
    
    // 实现自定义的SourceFunction
    public static class MySensorSource implements SourceFunction<SensorReading>{ 
        // 定义一个标识位，用来控制数据的产生
        private boolean running = true;
    
        @Override
        public void run(SourceContext<SensorReading> ctx) throws Exception { 
            // 定义一个随机数发生器
            Random random = new Random();
    
            // 设置10个传感器的初始温度
           HashMap<String, Double> sensorTempMap = new HashMap<>();
               for(int i = 0; i < 10; i++){ 
                  sensorTempMap.put("sensor_" + (i + 1), 60 + random.nextGaussian() * 20);
                }
                while (running){ 
                    for(String sensorId: sensorTempMap.keySet()){ 
                        // 在当前温度基础上随机波动
                        Double newtemp = sensorTempMap.get(sensorId) + random.nextGaussian();
                        sensorTempMap.put(sensorId, newtemp);
                        ctx.collect(new SensorReading(sensorId, System.currentTimeMillis(), newtemp));
                    }
                    // 控制输出频率
                    Thread.sleep(1000);
             }
        }
    
        @Override
        public void cancel() { 
    
        }
    }

## 三、Transform（转换算子） ##

### 3.1 基本转换算子 ###

#### 3.1.1 map ####

map算子属于来一个数据处理一个

#### 3.1.2 flatMap ####

flatMap算子常用于打散数据/拆分数据

#### 3.1.3 filter ####

filter算子顾名思义用于过滤数据，结果数据的类型与原始数据相同

示例代码：

public class TransformTest1_Base { 
        public static void main(String[] args) throws Exception{ 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // 从文件中读取数据
            DataStream<String> inputStream = env.readTextFile("/Users/sundongping/IdeaProjects/FirstFlinkTest/src/main/resources/sensor.txt");
            // 1. map 把String转换成长度输出
            DataStream<Integer> mapStream = inputStream.map(new MapFunction<String, Integer>() { 
                @Override
                public Integer map(String s) throws Exception { 
                    return s.length();
                }
            });
    
            // 2. flatMap 按逗号切分字段
            DataStream<String> flatMapStream = inputStream.flatMap(new FlatMapIterator<String, String>() { 
                @Override
                public Iterator<String> flatMap(String s) throws Exception { 
                    List<String> arrayList = new ArrayList<>();
    
                    String[] fields = s.split(",");
                    for(String field : fields){ 
                        arrayList.add(field);
                    }
                    return arrayList.iterator();
                }
            });
    
            // 3. filter 筛选sensor_1开头的id对应的数据
            DataStream<String> filterStream = inputStream.filter(new FilterFunction<String>() { 
                @Override
                public boolean filter(String s) throws Exception { 
                    return s.startsWith("sensor_1");
                }
            });
    
            // 打印输出
            mapStream.print("map");
            flatMapStream.print("flatMap");
            filterStream.print("filter");
    
            env.execute();
        }
    }

运行结果：

map> 27
    flatMap> sensor_1
    flatMap> 154777324345L
    flatMap> 35.8
    filter> sensor_1,154777324345L,35.8
    map> 27
    flatMap> sensor_2
    flatMap> 154777324346L
    flatMap> 15.4
    map> 26
    flatMap> sensor_3
    flatMap> 154777324347L
    flatMap> 6.7
    map> 27
    flatMap> sensor_4
    flatMap> 154777324348L
    flatMap> 38.1

### 3.2 聚合相关算子 ###

### KeyBy ###

DataStream -> KeyedStream：逻辑地将一个流拆分成不相交的分区，每个分区包含具有相同key的元素，在内部以hash的形式实现的

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

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

*  sum()
 *  min()
 *  max()
 *  minBy()
 *  maxBy()

max()示例：

public class TransformTest2_RollingAggregation { 
        public static void main(String[] args) throws Exception{ 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // 从文件中读取数据
            DataStream<String> inputStream = env.readTextFile("/Users/sundongping/IdeaProjects/FirstFlinkTest/src/main/resources/sensor.txt");
    
            // 转换成SensorReading类型
    //        DataStream<SensorReading> dataStream = inputStream.map(new MapFunction<String, SensorReading>() { 
    //            @Override
    //            public SensorReading map(String s) throws Exception { 
    //                String[] fields = s.split(",");
    //                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
    //            }
    //        });
            // java8 中的lamda表达式
            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");
            // java8 方法调用
            KeyedStream<SensorReading, String> keyedStream = dataStream.keyBy(data -> data.getId());
            // java8 方法引用
    //        KeyedStream<SensorReading, String> keyedStream1 = dataStream.keyBy(SensorReading::getId);
    
            // 滚动聚合，取当前最大的温度值
            SingleOutputStreamOperator<SensorReading> resultStream = keyedStream.max("temperature");
    
            // 打印输出
            resultStream.print();
    
            env.execute();
        }
    }

运行结果：  
只有最大温度字段变了，其他字段值不变  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70]

maxBy()示例：

public class TransformTest2_RollingAggregation { 
        public static void main(String[] args) throws Exception{ 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // 从文件中读取数据
            DataStream<String> inputStream = env.readTextFile("/Users/sundongping/IdeaProjects/FirstFlinkTest/src/main/resources/sensor.txt");
    
            // 转换成SensorReading类型
    //        DataStream<SensorReading> dataStream = inputStream.map(new MapFunction<String, SensorReading>() { 
    //            @Override
    //            public SensorReading map(String s) throws Exception { 
    //                String[] fields = s.split(",");
    //                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
    //            }
    //        });
            // java8 中的lamda表达式
            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");
            // java8 方法调用
            KeyedStream<SensorReading, String> keyedStream = dataStream.keyBy(data -> data.getId());
            // java8 方法引用
    //        KeyedStream<SensorReading, String> keyedStream1 = dataStream.keyBy(SensorReading::getId);
    
            // 滚动聚合，取当前最大的温度值
            SingleOutputStreamOperator<SensorReading> resultStream = keyedStream.maxBy("temperature");
    
            // 打印输出
            resultStream.print();
    
            env.execute();
        }
    }

运行结果：  
最大温度值变了，其他字段也会随之改变，变成最大温度值所属对象的  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 1]

#### 3.2.2 Reduce ####

KeyedStream -> DataStream：一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果

public class TransformTest3_Reduce { 
        public static void main(String[] args) throws Exception { 
            
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // 从文件中读取数据
            DataStream<String> inputStream = env.readTextFile("/Users/sundongping/IdeaProjects/FirstFlinkTest/src/main/resources/sensor.txt");
    
            // 转换成SensorReading类型
            // java8 中的lamda表达式
            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聚合，取最大的温度值，以及当前最新的时间戳
    //        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()));
    //            }
    //        });
            
            // java8 lamda表达式
            SingleOutputStreamOperator<SensorReading> resultStr = keyedStream.reduce((curState, newData) -> { return new SensorReading(curState.getId(), newData.getTimestamp(), Math.max(curState.getTemperature(), newData.getTemperature()));
            });
    
            resultStr.print();
    
            env.execute();
        }
    }

运行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 2]

### 3.3 多流转换算子 ###

#### 3.3.1 Split 和 Select（组合使用） ####

Split算子用于按一定特征把一条流切成两条流或多条流。DataStream -> SplitStream：根据某些特征把一个 DataStream 拆分成两个或者多个 SplitStream  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 3]  
Select算子  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 4]  
示例代码：

运行结果：

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
    SLF4J: Defaulting to no-operation (NOP) logger implementation
    SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
    all> SensorReading{ id='sensor_1', timestamp=154777324345, temperature=35.8}
    high> SensorReading{ id='sensor_1', timestamp=154777324345, temperature=35.8}
    low> SensorReading{ id='sensor_2', timestamp=154777324346, temperature=15.4}
    all> SensorReading{ id='sensor_2', timestamp=154777324346, temperature=15.4}
    low> SensorReading{ id='sensor_3', timestamp=154777324347, temperature=6.7}
    all> SensorReading{ id='sensor_3', timestamp=154777324347, temperature=6.7}
    all> SensorReading{ id='sensor_4', timestamp=154777324348, temperature=38.1}
    high> SensorReading{ id='sensor_4', timestamp=154777324348, temperature=38.1}
    all> SensorReading{ id='sensor_1', timestamp=154777324349, temperature=36.9}
    high> SensorReading{ id='sensor_1', timestamp=154777324349, temperature=36.9}
    all> SensorReading{ id='sensor_1', timestamp=154777324350, temperature=32.9}
    high> SensorReading{ id='sensor_1', timestamp=154777324350, temperature=32.9}
    all> SensorReading{ id='sensor_1', timestamp=154777324351, temperature=37.1}
    high> SensorReading{ id='sensor_1', timestamp=154777324351, temperature=37.1}
    low> SensorReading{ id='sensor_6', timestamp=154777324351, temperature=15.4}
    all> SensorReading{ id='sensor_6', timestamp=154777324351, temperature=15.4}
    low> SensorReading{ id='sensor_7', timestamp=154777324350, temperature=6.7}
    all> SensorReading{ id='sensor_7', timestamp=154777324350, temperature=6.7}

#### 3.3.2 Connect 和 CoMap（组合使用） ####

Connect算子，DataStream -> ConnectedStreams：连接两个保持他们类型的数据流，两个数据流被Connect之后，只是被放在了同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 5]  
CoMap,CoFlatMap算子，ConnectedStreams -> DataStream：作用于 ConnectedStreams 上，功能与map 和 flatMap一样，对 ConnectedStreams 中的每一个 Stream 分别进行 map和flatMap 处理  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 6]  
示例代码：

public class TransformTest4_MultiplieStreams { 
        public static void main(String[] args) throws Exception { 
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // 从文件中读取数据
            DataStream<String> inputStream = env.readTextFile("/Users/sundongping/IdeaProjects/FirstFlinkTest/src/main/resources/sensor.txt");
    
            // 转换成SensorReading类型
            // java8 中的lamda表达式
            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度为界分为两条流，split已被弃用
            SplitStream<SensorReading> splitStream = dataStream.split(new OutputSelector<SensorReading>() { 
                @Override
                public Iterable<String> select(SensorReading sensorReading) { 
                    return (sensorReading.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 sensorReading) throws Exception { 
                    return new Tuple2<>(sensorReading.getId(), sensorReading.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();
    
            env.execute();
        }
    }

运行结果：

high> SensorReading{ id='sensor_1', timestamp=154777324345, temperature=35.8}
    all> SensorReading{ id='sensor_1', timestamp=154777324345, temperature=35.8}
    low> SensorReading{ id='sensor_2', timestamp=154777324346, temperature=15.4}
    all> SensorReading{ id='sensor_2', timestamp=154777324346, temperature=15.4}
    low> SensorReading{ id='sensor_3', timestamp=154777324347, temperature=6.7}
    all> SensorReading{ id='sensor_3', timestamp=154777324347, temperature=6.7}
    high> SensorReading{ id='sensor_4', timestamp=154777324348, temperature=38.1}
    all> SensorReading{ id='sensor_4', timestamp=154777324348, temperature=38.1}
    high> SensorReading{ id='sensor_1', timestamp=154777324349, temperature=36.9}
    all> SensorReading{ id='sensor_1', timestamp=154777324349, temperature=36.9}
    high> SensorReading{ id='sensor_1', timestamp=154777324350, temperature=32.9}
    all> SensorReading{ id='sensor_1', timestamp=154777324350, temperature=32.9}
    high> SensorReading{ id='sensor_1', timestamp=154777324351, temperature=37.1}
    all> SensorReading{ id='sensor_1', timestamp=154777324351, temperature=37.1}
    low> SensorReading{ id='sensor_6', timestamp=154777324351, temperature=15.4}
    all> SensorReading{ id='sensor_6', timestamp=154777324351, temperature=15.4}
    low> SensorReading{ id='sensor_7', timestamp=154777324350, temperature=6.7}
    all> SensorReading{ id='sensor_7', timestamp=154777324350, temperature=6.7}
    (sensor_1,35.8,high temp warning)
    (sensor_2,normal)
    (sensor_4,38.1,high temp warning)
    (sensor_3,normal)
    (sensor_1,36.9,high temp warning)
    (sensor_6,normal)
    (sensor_1,32.9,high temp warning)
    (sensor_7,normal)
    (sensor_1,37.1,high temp warning)

缺点：只能针对两条流，多条流不能使用

#### 3.3.3 Union ####

DataStream -> DataStream：对两个或者两个以上**类型相同**的 DataStream 进行 union 操作，产生一个包含所有 DataStream 元素的新DataStream。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 7]  
示例代码：

运行结果:

high> SensorReading{ id='sensor_1', timestamp=154777324345, temperature=35.8}
    all> SensorReading{ id='sensor_1', timestamp=154777324345, temperature=35.8}
    all> SensorReading{ id='sensor_2', timestamp=154777324346, temperature=15.4}
    low> SensorReading{ id='sensor_2', timestamp=154777324346, temperature=15.4}
    all> SensorReading{ id='sensor_3', timestamp=154777324347, temperature=6.7}
    low> SensorReading{ id='sensor_3', timestamp=154777324347, temperature=6.7}
    high> SensorReading{ id='sensor_4', timestamp=154777324348, temperature=38.1}
    all> SensorReading{ id='sensor_4', timestamp=154777324348, temperature=38.1}
    high> SensorReading{ id='sensor_1', timestamp=154777324349, temperature=36.9}
    all> SensorReading{ id='sensor_1', timestamp=154777324349, temperature=36.9}
    high> SensorReading{ id='sensor_1', timestamp=154777324350, temperature=32.9}
    all> SensorReading{ id='sensor_1', timestamp=154777324350, temperature=32.9}
    high> SensorReading{ id='sensor_1', timestamp=154777324351, temperature=37.1}
    all> SensorReading{ id='sensor_1', timestamp=154777324351, temperature=37.1}
    all> SensorReading{ id='sensor_6', timestamp=154777324351, temperature=15.4}
    low> SensorReading{ id='sensor_6', timestamp=154777324351, temperature=15.4}
    all> SensorReading{ id='sensor_7', timestamp=154777324350, temperature=6.7}
    low> SensorReading{ id='sensor_7', timestamp=154777324350, temperature=6.7}
    union> SensorReading{ id='sensor_1', timestamp=154777324345, temperature=35.8}
    union> SensorReading{ id='sensor_4', timestamp=154777324348, temperature=38.1}
    union> SensorReading{ id='sensor_1', timestamp=154777324349, temperature=36.9}
    union> SensorReading{ id='sensor_1', timestamp=154777324350, temperature=32.9}
    union> SensorReading{ id='sensor_1', timestamp=154777324351, temperature=37.1}
    union> SensorReading{ id='sensor_2', timestamp=154777324346, temperature=15.4}
    union> SensorReading{ id='sensor_3', timestamp=154777324347, temperature=6.7}
    union> SensorReading{ id='sensor_6', timestamp=154777324351, temperature=15.4}
    union> SensorReading{ id='sensor_7', timestamp=154777324350, temperature=6.7}
    (sensor_1,35.8,high temp warning)
    (sensor_2,normal)
    (sensor_4,38.1,high temp warning)
    (sensor_3,normal)
    (sensor_1,36.9,high temp warning)
    (sensor_6,normal)
    (sensor_1,32.9,high temp warning)
    (sensor_7,normal)
    (sensor_1,37.1,high temp warning)

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70]: /images/20221022/ec301df2f3c74886b6085548c22d426d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 1]: /images/20221022/71347b05dbab4abfb6c4e3f35abc089d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 2]: /images/20221022/5bf5f010c80b4180ab5583a88337c49a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 3]: /images/20221022/8c4b2be53b6c440a990f6c5ce91f2b33.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 4]: /images/20221022/f51e101b8c644e56ba831a8e623bde7e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 5]: /images/20221022/96e20dfc02464294b2424341268140a5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 6]: /images/20221022/4c1faeef0b6a46b4a371af17562bfda1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 7]: /images/20221022/a82fc2295b11438b97462d0b449233f6.png