Open ChenShuai1981 opened 4 years ago
我将WSO2 Siddhi repo下的示例 TimeWindowSample 用 flink-siddhi 库改写并运行,发现和原生的示例输出结果并不相同,难道是我哪里写得不对吗?
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); DataStream<Tuple4<String, Float, Long, Long>> stockStream = env.fromElements( Tuple4.of("IBM", 100f, 100L, 1000L), Tuple4.of("IBM", 200f, 300L, 2000L), Tuple4.of("WSO2", 60f, 200L, 2000L), Tuple4.of("WSO2", 70f, 400L, 3000L), Tuple4.of("GOOG", 50f, 30L, 3000L), Tuple4.of("IBM", 200f, 400L, 4000L), Tuple4.of("WSO2", 70f, 50L, 6000L), Tuple4.of("WSO2", 80f, 400L, 8000L), Tuple4.of("GOOG", 60f, 30L, 8000L) ); SiddhiCEP cep = SiddhiCEP.getSiddhiEnvironment(env); SiddhiStream.SingleSiddhiStream stream = cep.define("StockStream", stockStream, "symbol", "price", "volume", "timestamp"); DataStream output = stream .cql("from StockStream#window.time(5 sec) select symbol, sum(price) as price, sum(volume) as volume group by symbol insert into AggregateStockStream") .returnsTransformRow("AggregateStockStream"); output.print(); env.execute();
输出结果是
IBM,100.0,100 GOOG,60.0,30 WSO2,80.0,400 WSO2,150.0,450 IBM,300.0,500 GOOG,110.0,60 WSO2,220.0,850 WSO2,280.0,1050 IBM,500.0,800
原生输出结果是
[Event{timestamp=1576396964357, data=[IBM, 100.0, 100], isExpired=false}] [Event{timestamp=1576396965368, data=[IBM, 300.0, 400], isExpired=false}] [Event{timestamp=1576396965369, data=[WSO2, 60.0, 200], isExpired=false}] [Event{timestamp=1576396966373, data=[WSO2, 130.0, 600], isExpired=false}] [Event{timestamp=1576396966373, data=[GOOG, 50.0, 30], isExpired=false}] [Event{timestamp=1576396967373, data=[IBM, 500.0, 800], isExpired=false}] [Event{timestamp=1576396969375, data=[WSO2, 200.0, 650], isExpired=false}] [Event{timestamp=1576396971380, data=[WSO2, 150.0, 450], isExpired=false}] [Event{timestamp=1576396971380, data=[GOOG, 60.0, 30], isExpired=false}]
难道Siddhi只支持processing time处理时间吗?
我测试了一下,你把env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)这个删了就一样了
我将WSO2 Siddhi repo下的示例 TimeWindowSample 用 flink-siddhi 库改写并运行,发现和原生的示例输出结果并不相同,难道是我哪里写得不对吗?
输出结果是
原生输出结果是