Hi clay, Are there any other lines after the last line in your picture? The final result should be eventual consistency and correct.
In your sql, there is a left join, a keyed group by and a non-keyed group by. Both of the left join and keyed group by will send retractions to the downstream non-keyed group by once there is an update. The retraction messages vibrate the result value. However, the final result will be correct. To get monotonous results, you can add another non-keyed group by with max. Best, Hequn. On Sat, Sep 29, 2018 at 3:47 PM clay4444 <clay4me...@gmail.com> wrote: > My final calculation result is implemented in the following way when > writing > to kafka, because KafkaTableSink does not support retract mode, I am not > sure whether this method will affect the calculation result. > > val userTest: Table = tEnv.sqlQuery(sql) > > val endStream = tEnv.toRetractStream[Row](userTest) > > //userTest.insertInto("kafkaSink") > > val myProducer = new FlinkKafkaProducer011[String]( > kafkaBrokers, // broker list > topic, // target topic > new SimpleStringSchema) // serialization schema > > endStream.map(x=>{ > s"${x._1}:${x._2.toString}" > }).addSink(myProducer) > > > > -- > Sent from: > http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/ >