Ah, that's great! Thanks for letting us know :-) Am Mo., 24. Juni 2019 um 11:33 Uhr schrieb Wouter Zorgdrager < w.d.zorgdra...@tudelft.nl>:
> Hi Fabian, > > Thanks for your reply. I managed to resolve this issue. Actually this > behavior was not so unexpected, I messed up using xStream as a 'base' while > I needed to use yStream as a 'base'. I.e. yStream.element - 60 min <= > xStream.element <= yStream.element + 30 min. Interchanging both datastreams > fixed this issue. > > Thanks anyways. > > Cheers, Wouter > > > > Op ma 24 jun. 2019 om 11:22 schreef Fabian Hueske <fhue...@gmail.com>: > >> Hi Wouter, >> >> Not sure what is going wrong there, but something that you could try is >> to use a custom watemark assigner and always return a watermark of 0. >> When the source finished serving the watermarks, it emits a final >> Long.MAX_VALUE watermark. >> Hence the join should consume all events and store them in state. When >> both sources are finished, it would start to join the data and clean up the >> state. >> This test would show if there are any issue with late data. >> >> Best, Fabian >> >> Am Fr., 21. Juni 2019 um 15:32 Uhr schrieb Wouter Zorgdrager < >> w.d.zorgdra...@tudelft.nl>: >> >>> Anyone some leads on this issue? Have been looking into the >>> IntervalJoinOperator code, but that didn't really help. My intuition is >>> that it is rejected because of lateness, however that still confuses me >>> since I'm sure that both datastreams have monotonic increasing timestamps. >>> >>> Thx, Wouter >>> >>> Op ma 17 jun. 2019 om 13:20 schreef Wouter Zorgdrager < >>> w.d.zorgdra...@tudelft.nl>: >>> >>>> Hi all, >>>> >>>> I'm experiencing some unexpected behavior using an interval join in >>>> Flink. >>>> I'm dealing with two data sets, lets call them X and Y. They are finite >>>> (10k elements) but I interpret them as a DataStream. The data needs to be >>>> joined for enrichment purposes. I use event time and I know (because I >>>> generated the data myself) that the timestamp of an element Y is always >>>> between -60 minutes and +30 minutes of the element with the same key in set >>>> X. Both datasets are in-order (in terms of timestamps), equal in size, >>>> share a common key and parallelism is set to 1 throughout the whole >>>> program. >>>> >>>> The code to join looks something like this: >>>> >>>> xStream >>>> .assignAscendingTimestamps(_.date.getTime) >>>> .keyBy(_.commonKey) >>>> .intervalJoin( >>>> yStream >>>> .assignAscendingTimestamps(_.date.getTime) >>>> .keyBy(_.commonKey)) >>>> .between(Time.minutes(-60), Time.minutes(30)) >>>> .process(new ProcessJoinFunction[X, Y, String] { >>>> override def processElement( >>>> left: X, >>>> right: Y, >>>> ctx: ProcessJoinFunction[X, Y, String]#Context, >>>> out: Collector[String]): Unit = { >>>> >>>> out.collect(left + ":" + right) >>>> } >>>> >>>> >>>> However, about 30% percent of the data is not joined. Is there a proper >>>> way to debug this? For instance, in windows you can side-output late data. >>>> Is there a possibility to side-output unjoinable data? >>>> >>>> Thx a lot, >>>> Wouter >>>> >>>> >>>>