Hey Timo, Hah, that's a fair point about using time. I guess I should update my statement to "as a user, I don't want to worry about *manually managing* time".
That's a nice suggestion with the KeyedProcessFunction and no windows, I'll give that a shot. If I don't want to emit any duplicates, I'd have to essentially buffer the "last seen duplicate" for each key in that process function until the MAX_WATERMARK is sent through though, right? I could emit early results if I assume the max number of possible duplicates, but for records with no duplicates, I'd have to wait until no more records are coming -- am I missing something? Thanks so much, Austin On Fri, Oct 9, 2020 at 10:44 AM Timo Walther <twal...@apache.org> wrote: > Hi Austin, > > if you don't want to worry about time at all, you should probably not > use any windows because those are a time-based operation. > > A solution that would look a bit nicer could be to use a pure > KeyedProcessFunction and implement the deduplication logic without > reusing windows. In ProcessFunctions you can register an event-time > timer. The timer would be triggered by the MAX_WATERMARK when the > pipeline shuts down even without having a timestamp assigned in the > StreamRecord. Watermark will leave SQL also without a time attribute as > far as I know. > > Regards, > Timo > > > On 08.10.20 17:38, Austin Cawley-Edwards wrote: > > Hey Timo, > > > > Sorry for the delayed reply. I'm using the Blink planner and using > > non-time-based joins. I've got an example repo here that shows my query/ > > setup [1]. It's got the manual timestamp assignment commented out for > > now, but that does indeed solve the issue. > > > > I'd really like to not worry about time at all in this job hah -- I > > started just using processing time, but Till pointed out that processing > > time timers won't be fired when input ends, which is the case for this > > streaming job processing CSV files, so I should be using event time. > > With that suggestion, I switched to ingestion time, where I then > > discovered the issue converting from SQL to data stream. > > > > IMO, as a user manually assigning timestamps on conversion makes sense > > if you're using event time and already handling time attributes > > yourself, but for ingestion time you really don't want to think about > > time at all, which is why it might make sense to propigate the > > automatically assigned timestamps in that case. Though not sure how > > difficult that would be. Let me know what you think! > > > > > > Best + thanks again, > > Austin > > > > [1]: https://github.com/austince/flink-1.10-sql-windowing-error > > > > On Mon, Oct 5, 2020 at 4:24 AM Timo Walther <twal...@apache.org > > <mailto:twal...@apache.org>> wrote: > > > > Btw which planner are you using? > > > > Regards, > > Timo > > > > On 05.10.20 10:23, Timo Walther wrote: > > > Hi Austin, > > > > > > could you share some details of your SQL query with us? The > > reason why > > > I'm asking is because I guess that the rowtime field is not > inserted > > > into the `StreamRecord` of DataStream API. The rowtime field is > only > > > inserted if there is a single field in the output of the query > > that is a > > > valid "time attribute". > > > > > > Esp. after non-time-based joins and aggregations, time attributes > > loose > > > there properties and become regular timestamps. Because timestamp > > and > > > watermarks might have diverged. > > > > > > If you know what you're doing, you can also assign the timestamp > > > manually after `toRetractStream.assignTimestampAndWatermarks` and > > > reinsert the field into the stream record. But before you do > that, I > > > think it is better to share more information about the query with > us. > > > > > > I hope this helps. > > > > > > Regards, > > > Timo > > > > > > > > > > > > On 05.10.20 09:25, Till Rohrmann wrote: > > >> Hi Austin, > > >> > > >> thanks for offering to help. First I would suggest asking Timo > > whether > > >> this is an aspect which is still missing or whether we > > overlooked it. > > >> Based on that we can then take the next steps. > > >> > > >> Cheers, > > >> Till > > >> > > >> On Fri, Oct 2, 2020 at 7:05 PM Austin Cawley-Edwards > > >> <austin.caw...@gmail.com <mailto:austin.caw...@gmail.com> > > <mailto:austin.caw...@gmail.com <mailto:austin.caw...@gmail.com>>> > > wrote: > > >> > > >> Hey Till, > > >> > > >> Thanks for the notes. Yeah, the docs don't mention anything > > specific > > >> to this case, not sure if it's an uncommon one. Assigning > > timestamps > > >> on conversion does solve the issue. I'm happy to take a stab > at > > >> implementing the feature if it is indeed missing and you all > > think > > >> it'd be worthwhile. I think it's definitely a confusing > > aspect of > > >> working w/ the Table & DataStream APIs together. > > >> > > >> Best, > > >> Austin > > >> > > >> On Fri, Oct 2, 2020 at 6:05 AM Till Rohrmann > > <trohrm...@apache.org <mailto:trohrm...@apache.org> > > >> <mailto:trohrm...@apache.org <mailto:trohrm...@apache.org>>> > > wrote: > > >> > > >> Hi Austin, > > >> > > >> yes, it should also work for ingestion time. > > >> > > >> I am not entirely sure whether event time is preserved > when > > >> converting a Table into a retract stream. It should be > > possible > > >> and if it is not working, then I guess it is a missing > > feature. > > >> But I am sure that @Timo Walther > > >> <mailto:twal...@apache.org > > <mailto:twal...@apache.org>> knows more about it. In doubt, you > > >> could assign a new watermark generator when having > > obtained the > > >> retract stream. > > >> > > >> Here is also a link to some information about event time > and > > >> watermarks [1]. Unfortunately, it does not state > > anything about > > >> the direction Table => DataStream. > > >> > > >> [1] > > >> > > >> > > > https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/streaming/time_attributes.html > > > > >> > > >> > > >> Cheers, > > >> Till > > >> > > >> On Fri, Oct 2, 2020 at 12:10 AM Austin Cawley-Edwards > > >> <austin.caw...@gmail.com > > <mailto:austin.caw...@gmail.com> <mailto:austin.caw...@gmail.com > > <mailto:austin.caw...@gmail.com>>> wrote: > > >> > > >> Hey Till, > > >> > > >> Just a quick question on time characteristics -- > > this should > > >> work for IngestionTime as well, correct? Is there > > anything > > >> special I need to do to have the CsvTableSource/ > > >> toRetractStream call to carry through the assigned > > >> timestamps, or do I have to re-assign timestamps > > during the > > >> conversion? I'm currently getting the `Record has > > >> Long.MIN_VALUE timestamp (= no timestamp marker)` > error, > > >> though I'm seeing timestamps being assigned if I step > > >> through the AutomaticWatermarkContext. > > >> > > >> Thanks, > > >> Austin > > >> > > >> On Thu, Oct 1, 2020 at 10:52 AM Austin Cawley-Edwards > > >> <austin.caw...@gmail.com > > <mailto:austin.caw...@gmail.com> <mailto:austin.caw...@gmail.com > > <mailto:austin.caw...@gmail.com>>> > > >> wrote: > > >> > > >> Perfect, thanks so much Till! > > >> > > >> On Thu, Oct 1, 2020 at 5:13 AM Till Rohrmann > > >> <trohrm...@apache.org > > <mailto:trohrm...@apache.org> <mailto:trohrm...@apache.org > > <mailto:trohrm...@apache.org>>> > > >> wrote: > > >> > > >> Hi Austin, > > >> > > >> I believe that the problem is the processing > > time > > >> window. Unlike for event time where we send a > > >> MAX_WATERMARK at the end of the stream to > > trigger > > >> all remaining windows, this does not happen > for > > >> processing time windows. Hence, if your > > stream ends > > >> and you still have an open processing time > > window, > > >> then it will never get triggered. > > >> > > >> The problem should disappear if you use > > event time > > >> or if you process unbounded streams which > > never end. > > >> > > >> Cheers, > > >> Till > > >> > > >> On Thu, Oct 1, 2020 at 12:01 AM Austin > > >> Cawley-Edwards <austin.caw...@gmail.com > > <mailto:austin.caw...@gmail.com> > > >> <mailto:austin.caw...@gmail.com > > <mailto:austin.caw...@gmail.com>>> wrote: > > >> > > >> Hey all, > > >> > > >> Thanks for your patience. I've got a > > small repo > > >> that reproduces the issue here: > > >> > > >> https://github.com/austince/flink-1.10-sql-windowing-error > > >> > > >> > > >> Not sure what I'm doing wrong but it > > feels silly. > > >> > > >> Thanks so much! > > >> Austin > > >> > > >> On Tue, Sep 29, 2020 at 3:48 PM Austin > > >> Cawley-Edwards <austin.caw...@gmail.com > > <mailto:austin.caw...@gmail.com> > > >> <mailto:austin.caw...@gmail.com > > <mailto:austin.caw...@gmail.com>>> wrote: > > >> > > >> Hey Till, > > >> > > >> Thanks for the reply -- I'll try to > > see if I > > >> can reproduce this in a small repo > > and share > > >> it with you. > > >> > > >> Best, > > >> Austin > > >> > > >> On Tue, Sep 29, 2020 at 3:58 AM Till > > >> Rohrmann <trohrm...@apache.org > > <mailto:trohrm...@apache.org> > > >> <mailto:trohrm...@apache.org > > <mailto:trohrm...@apache.org>>> wrote: > > >> > > >> Hi Austin, > > >> > > >> could you share with us the > > exact job > > >> you are running (including the > > custom > > >> window trigger)? This would help > > us to > > >> better understand your problem. > > >> > > >> I am also pulling in Klou and > > Timo who > > >> might help with the windowing > > logic and > > >> the Table to DataStream > conversion. > > >> > > >> Cheers, > > >> Till > > >> > > >> On Mon, Sep 28, 2020 at 11:49 PM > > Austin > > >> Cawley-Edwards > > <austin.caw...@gmail.com <mailto:austin.caw...@gmail.com> > > >> <mailto:austin.caw...@gmail.com > > <mailto:austin.caw...@gmail.com>>> wrote: > > >> > > >> Hey all, > > >> > > >> I'm not sure if I've missed > > >> something in the docs, but > I'm > > >> having a bit of trouble with > a > > >> streaming SQL job that > > starts w/ raw > > >> SQL queries and then > > transitions to > > >> a more traditional streaming > > job. > > >> I'm on Flink 1.10 using the > > Blink > > >> planner, running locally > with no > > >> checkpointing. > > >> > > >> The job looks roughly like: > > >> > > >> CSV 1 --> > > >> CSV 2 --> SQL Query to Join > --> > > >> toRetractStream --> keyed > time > > >> window w/ process func & > custom > > >> trigger --> some other ops > > >> CSV 3 --> > > >> > > >> > > >> When I remove the windowing > > directly > > >> after the `toRetractStream`, > the > > >> records make it to the "some > > other > > >> ops" stage, but with the > > windowing, > > >> those operations are > > sometimes not > > >> sent any data. I can also > > get data > > >> sent to the downstream > > operators by > > >> putting in a no-op map > > before the > > >> window and placing some > > breakpoints > > >> in there to manually slow > down > > >> processing. > > >> > > >> > > >> The logs don't seem to > indicate > > >> anything went wrong and > > generally > > >> look like: > > >> > > >> 4819 [Source: Custom File > source > > >> (1/1)] INFO > > >> > > >> org.apache.flink.runtime.taskmanager.Task - Source: Custom File > > >> source (1/1) (3578629787c777320d9ab030c004abd4) switched from > > RUNNING > > >> to FINISHED.\4819 [Source: Custom File source (1/1)] INFO > > >> org.apache.flink.runtime.taskmanager.Task - Freeing task > > resources > > >> for Source: Custom File source (1/1) > > (3578629787c777320d9ab030c004abd4). > > >> 4819 [Source: Custom File > source > > >> (1/1)] INFO > > >> > > >> org.apache.flink.runtime.taskmanager.Task - Ensuring all > > FileSystem > > >> streams are closed for task Source: Custom File source (1/1) > > >> (3578629787c777320d9ab030c004abd4) [FINISHED] > > >> 4820 > > >> > > >> [flink-akka.actor.default-dispatcher-5] > > >> INFO > > >> > > >> org.apache.flink.runtime.taskexecutor.TaskExecutor - > > Un-registering > > >> task and sending final execution state FINISHED to JobManager > > for task > > >> Source: Custom File source (1/1) > 3578629787c777320d9ab030c004abd4. > > >> ... > > >> 4996 > > >> > > >> [Window(TumblingProcessingTimeWindows(60000), > > >> TimedCountTrigger, > > >> ProcessWindowFunction$1) > > (1/1)] INFO > > >> > > >> org.apache.flink.runtime.taskmanager.Task - > > >> Window(TumblingProcessingTimeWindows(60000), TimedCountTrigger, > > >> ProcessWindowFunction$1) (1/1) (907acf9bfa2f4a9bbd13997b8b34d91f) > > >> switched from RUNNING to FINISHED. > > >> 4996 > > >> > > >> [Window(TumblingProcessingTimeWindows(60000), > > >> TimedCountTrigger, > > >> ProcessWindowFunction$1) > > (1/1)] INFO > > >> > > >> org.apache.flink.runtime.taskmanager.Task - Freeing task > > resources > > >> for Window(TumblingProcessingTimeWindows(60000), > TimedCountTrigger, > > >> ProcessWindowFunction$1) (1/1) > (907acf9bfa2f4a9bbd13997b8b34d91f). > > >> 4996 > > >> > > >> [Window(TumblingProcessingTimeWindows(60000), > > >> TimedCountTrigger, > > >> ProcessWindowFunction$1) > > (1/1)] INFO > > >> > > >> org.apache.flink.runtime.taskmanager.Task - Ensuring all > > FileSystem > > >> streams are closed for task > > >> Window(TumblingProcessingTimeWindows(60000), TimedCountTrigger, > > >> ProcessWindowFunction$1) (1/1) (907acf9bfa2f4a9bbd13997b8b34d91f) > > >> [FINISHED] > > >> ... > > >> rest of the shutdown > > >> ... > > >> Program execution finished > > >> Job with JobID > > >> > > 889b161e432c0e69a8d760bbed205d5d has > > >> finished. > > >> Job Runtime: 783 ms > > >> > > >> > > >> Is there something I'm > > missing in my > > >> setup? Could it be my custom > > window > > >> trigger? Bug? I'm stumped. > > >> > > >> > > >> Thanks, > > >> Austin > > >> > > >> > > >> > > >> > > >> > > >> > > >> > > >> > > >> > > >> > > >> > > >> > > > > > > >
