FLINK基础（99）: DS算子与窗口（10）多流转换算子(1) Union

UNION

DataStream.union()方法将两条或者多条DataStream合并成一条具有与输入流相同类型的输出DataStream。接下来的转换算子将会处理输入流中的所有元素。图5-5展示了union操作符如何将黑色和白色的事件流合并成一个单一输出流。

事件合流的方式为FIFO方式。操作符并不会产生一个特定顺序的事件流。union操作符也不会进行去重。每一个输入事件都被发送到了下一个操作符。

说明：

1.union 合并的流的元素必须是相同的

2.union 可以合并多条流

3.union不去重，合流顺序为先进先出

实例一：

下面的例子展示了如何将三条类型为SensorReading的数据流合并成一条流。

scala version

val parisStream: DataStream[SensorReading] = ...
val tokyoStream: DataStream[SensorReading] = ...
val rioStream: DataStream[SensorReading] = ...
val allCities: DataStream[SensorReading] = parisStream
  .union(tokyoStream, rioStream)

java version

DataStream<SensorReading> parisStream = ...
DataStream<SensorReading> tokyoStream = ...
DataStream<SensorReading> rioStream = ...
DataStream<SensorReading> allCities = parisStream
  .union(tokyoStream, rioStream)

实例二：

需求

本篇文章我们看一下union的用法，将多个币种的汇率流合并成一个汇率流。

解决方案

    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //USD -> CNY 汇率流
        DataStreamSource<ExchangeRateInfo> usdToCny = env.addSource(new ExchangeRateDataSource(CurrencyType.USD, CurrencyType.CNY, 7, 6),"USD-CNY");
        //EUR -> CNY 汇率流
        DataStreamSource<ExchangeRateInfo> eurToCny = env.addSource(new ExchangeRateDataSource(CurrencyType.EUR, CurrencyType.CNY, 8, 7),"EUR-CNY");
        //AUD -> CNY 汇率流
        DataStreamSource<ExchangeRateInfo> audToCny = env.addSource(new ExchangeRateDataSource(CurrencyType.AUD, CurrencyType.CNY, 5, 4),"AUD-CNY");
        //三个流合并为一个流
        DataStream<ExchangeRateInfo> allExchangeRate = usdToCny.union(eurToCny).union(audToCny);
        //将流标准输出
        allExchangeRate.print();

        env.execute("Flink Streaming Java API Skeleton");
    }

执行效果

通过union函数将
USD -> CNY 汇率流
EUR -> CNY 汇率流
AUD -> CNY 汇率流

统一合并成一个汇率流

1> ExchangeRateInfo [from=AUD, to=CNY, coefficient=4.60]
1> ExchangeRateInfo [from=USD, to=CNY, coefficient=6.74]
1> ExchangeRateInfo [from=EUR, to=CNY, coefficient=7.05]
2> ExchangeRateInfo [from=AUD, to=CNY, coefficient=4.03]
2> ExchangeRateInfo [from=USD, to=CNY, coefficient=6.36]
2> ExchangeRateInfo [from=EUR, to=CNY, coefficient=7.83]
3> ExchangeRateInfo [from=AUD, to=CNY, coefficient=4.31]
3> ExchangeRateInfo [from=USD, to=CNY, coefficient=6.71]
3> ExchangeRateInfo [from=EUR, to=CNY, coefficient=7.79]
4> ExchangeRateInfo [from=AUD, to=CNY, coefficient=4.08]
4> ExchangeRateInfo [from=USD, to=CNY, coefficient=6.21]
4> ExchangeRateInfo [from=EUR, to=CNY, coefficient=7.46]

小结

在sql中union是将多个查询结果集合汇总成一个结果集合，而在流计算中类似的将多个流合并成一个流。

代码地址

https://github.com/chaoxxx/learn-flink-stream-api/blob/master/src/main/java/fun/cosmozhu/session14/main/StreamTest.java

作者：cosmozhu --90后的老父亲，专注于保护地球的程序员
个人网站：https://www.cosmozhu.fun
欢迎转载，转载时请注明出处。