Hi David, Thanks for the clarification. Should we update the watermark for join operator when there's a watermark arrived from one of the input streams even if the watermark from another input stream is not arrived yet?
Shunxin On Fri, Sep 16, 2016 at 10:59 AM, David Yan <da...@datatorrent.com> wrote: > Actually, that's not entirely true. Here are the points about the watermark > tuple generation of the join operator: > > 1) We keep the timestamp of the latest watermark for each input port > > 2) We keep another timestamp that is equal to minimum of all the timestamps > mentioned in (1). > > 3) Upon arrival of a watermark from an input port, we update the timestamp > mentioned in (1), and evaluate (2). If the value of (2) changes, we > generate the watermark tuple with the timestamp that is equal to the new > value of (2). > > 4) That means initially, the watermark is only generated when we have seen > a watermark for all input ports. And the fact that we take the smallest > timestamp in (2) means we only consider a window as late only if all input > streams say that particular window is late. > > David > > > On Fri, Sep 16, 2016 at 10:42 AM, Shunxin Lu <lushun...@gmail.com> wrote: > > > Hi Chinmay, > > > > Base on the discussion I had with David, and David please correct me if I > > am wrong, the watermark for Windowed Join Operator should be indeed > > depending on all the input streams. If a tuple is considered late for one > > input stream, it should also be considered late for the whole join > > operator. That's why in the AbstractWindowedJoinOperator, it always > selects > > the watermark with the smallest timestamp from all the latest watermarks > > coming from upstreams as its current watermark, so that it can make sure > > that it's always keeping the strictest watermark to eliminate late > tuples. > > > > Shunxin > > > > On Fri, Sep 16, 2016 at 10:02 AM, David Yan <da...@datatorrent.com> > wrote: > > > > > I think in theory, the watermark should be sent by the input operator > > since > > > the input should have the knowledge of the criteria of lateness since > it > > > can depend on many factors like the time of the day, the source of the > > data > > > (e.g. mobile data), that the WindowedOperator should in general make no > > > assumption about. > > > > > > However, I think it's possible to implement some kind of watermark > > > generation in the WindowedOperator itself if that knowledge is not > > > available from the input. It's actually already doing that if you call > > > the setFixedWatermark > > > method, which will generate a watermark tuple, with a timestamp that is > > > based on the derived time from the streaming window id, downstream for > > each > > > streaming window. It's possible to add the support of heuristic > watermark > > > generation as well and you're welcome to take that up. > > > > > > For the Windowed Join operator, the watermark generated for downstream > > > depends on the watermark arriving from each input stream, and it's not > > just > > > a simple propagate. Shunxin can comment more on this. > > > > > > David > > > > > > > > > On Thu, Sep 15, 2016 at 11:21 PM, Chinmay Kolhatkar < > chin...@apache.org> > > > wrote: > > > > > > > Hi All, > > > > > > > > I was looking at Windowed Operator APIs and have to mention they're > > > pretty > > > > nicely done. > > > > > > > > I have a question related to watermark generation. > > > > > > > > What I understood is that for completion of processing of an event > > window > > > > one has provision for sending of watermark tuple from some previous > > stage > > > > in the DAG. I want to know who should be doing that and when should > be > > it > > > > done. > > > > > > > > For e.g. I saw a PR of Windows Join Operator in apex-malhar and I > would > > > > like to use it in my application. Can someone give me an example of > > how a > > > > DAG will look like with this operator which has a stage which > generates > > > > watermark? And how should that stage decide on when to generate a > > > watermark > > > > tuple? > > > > > > > > -Chinmay. > > > > > > > > > >
