The purpose of watermark is to set a limitation on handling records due to state going infinity. In other cases (non-stateful operations), it is pretty normal to handle all of records even they're pretty late.
Btw, there was some comments regarding this: while Spark delegates to filter out late records in stateful operations for now, some of us (including me) think filtering out late records in earlier phase (source, or just after source) makes more sense. It just didn't come out as action, but I think it is still valid. https://github.com/apache/spark/pull/21617#issuecomment-400119049 If we move the phase of filtering out late records, what you would like to do may become the default behavior. This also means the output may be also changed for queries which use non-stateful operations, so it is not a trivial change and may require consensus like SPIP process. Thanks, Jungtaek Lim (HeartSaVioR) 2018년 9월 30일 (일) 오후 5:19, chandan prakash <chandanbaran...@gmail.com>님이 작성: > Interesting question. > I do not think without any aggregation operation/groupBy , watermark is > supported currently . > > Reason: > Watermark in Structured Streaming is used for limiting the size of state > needed to keep intermediate information in-memory. > And state only comes in picture in case of stateful processing. > Also in the code, it seems that filtering out records on basis of > watermark happen only in case of stateful operators > (statefulOperators.scala) > Have not tried running code though and would like to know if someone can > shed more light on this. > > Regards, > Chandan > > > On Sat, Sep 22, 2018 at 7:43 PM peay <p...@protonmail.com.invalid> wrote: > >> Hello, >> >> I am trying to use watermarking without aggregation, to filter out >> records that are just too late, instead of appending them to the output. My >> understanding is that aggregation is required for `withWatermark` to have >> any effect. Is that correct? >> >> I am looking for something along the lines of >> >> ``` >> df.withWatermark("ts", ...).filter(F.col("ts") <F.getCurrentWatermark()) >> ``` >> >> Is there any way to get the watermark value to achieve that? >> >> Thanks! >> > > > -- > Chandan Prakash > >
