Flink-1.9流计算开发：十五、join函数

public class StreamTest {
    private static final Logger LOG = LoggerFactory.getLogger(StreamTest.class);

    @SuppressWarnings("deprecation")
    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //CNY -> USD 汇率流
        SingleOutputStreamOperator<ExchangeRateInfo> usdToCny = env.addSource(new ExchangeRateDataSource(CurrencyType.CNY, CurrencyType.USD, 7, 6),"USD-CNY")
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<ExchangeRateInfo>(Time.milliseconds(100)) {
                    @Override
                    public long extractTimestamp(ExchangeRateInfo element) {
                        return element.getTimeStamp().getTime();
                    }
                });
        //订单流
        SingleOutputStreamOperator<OrderInfo> orderDs = env.addSource(new OrderDataSource())
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<OrderInfo>(Time.milliseconds(100)) {
                    @Override
                    public long extractTimestamp(OrderInfo element) {
                        return element.getTimeStamp().getTime();
                    }
                });

        //订单流inner join 汇率流
        orderDs.join(usdToCny)
        .where(new KeySelector<OrderInfo, CurrencyType>() {
            private static final long serialVersionUID = 1L;
            @Override
            public CurrencyType getKey(OrderInfo value) throws Exception {
                return value.getCurrencyType();
            }
        })
        .equalTo(new KeySelector<ExchangeRateInfo, CurrencyType>() {
            private static final long serialVersionUID = 1L;
            @Override
            public CurrencyType getKey(ExchangeRateInfo value) throws Exception {
                return value.getFrom();
            }
        })
        //转换20s窗口期内的订单金额为美元
        .window(TumblingEventTimeWindows.of(Time.seconds(10)))
        //将订单金额人民币转换为美元
        .apply(new JoinFunction<OrderInfo, ExchangeRateInfo, String>() {
            private static final long serialVersionUID = 1L;
            @Override
            public String join(OrderInfo first, ExchangeRateInfo second) throws Exception {
                return "$"+first.getTotalAmt().divide(second.getCoefficient(), 2,BigDecimal.ROUND_HALF_UP).toPlainString();
            }
        })
        .print();

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

3> $125.05
3> $625.79
3> $95.79
3> $587.56
3> $387.35
3> $278.63
3> $40.45
3> $578.36
3> $15.48
3> $236.96

select * from OrderData a,ExchangeRate b where a.CurrencyType = b.From;

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