Robert, Yun, and Theo, Thanks for the responses! I'm very much looking forward to upgrading once those changes are made.
I hacked this in myself but likely in a much less elegant way than Theo. For anybody who is curious - I extended the DateTimeBucketAssigner class and overrode the getBucketId method so that I could use the AWS SDK to directly write an empty success file whenever a new bucket was opened. I keep a Map of the last bucket that I've already written to, as well as the current bucket, so that I don't overwhelm S3 with requests (getBucketId appears to be called quite frequently). It's far from ideal but it's performing quite nicely and generally I'm not seeing any significant race conditions between files closing in the previous bucket and opening in the next. I of course only write the _SUCCESS file to the previous bucket not the current one. I will keep an eye on that ticket, and very much appreciate the effort! Warm regards, Peter On Tue, May 12, 2020 at 11:00 AM Theo Diefenthal < theo.diefent...@scoop-software.de> wrote: > Hi Yun, > > For me, that sounds quite nice. I implemented the same for my application > a few weeks ago, but of course tailored only to my app. > What I did: > 1. I wrapped the Parquet-StreamingFileSink into a Process-Function. > 2. I extended the default ProcessOperator and instead of > "notifyCheckpointComplete(long checkpointId)", I provided my > WrappedProcessFunction a "notifyCheckpointComplete(checkointId, > lastCommitWatermark)". > 3. I added a custom sink with parallelism 1 behind the > WrappedProcessFunction. > 4. From my WrappedProcessFunction, in notifyCheckpointComplete, I send a > message downstream to the parallelism 1 sink containing data about which > partitions were written to between in the phase to the last checkpoint. > 5. In the parallelism 1 sink, I make sure that I get the messages from all > upstream task (Give the constructor an int parameter telling it the > parallelism of the WrappedProcessFunction) and then perform my parallelism > 1 operation, in my case, telling Impala which partitions were added or got > new data. Luckily, in case of Impala, that operation can be made idempotent > so I only needed to make sure that I have an at least once processing from > the state perspective here. > > I had to go for notifyCheckpointComplete as only there, the parquet files > are ultimately committed and thus available for spark, impala and so on. > > So if you go on with that issue, I'd be really happy to be able to > customize the solution and e.g. get rid of my custom setup by only > specifiying kind of a lambda function which should be run with parallelism > 1 and update impala. That function would however still need the info which > partitions were updated/added. > And in my case, I was not really interested in the watermark (I sent it > downstream only for metric purposes) but want to tell impala after each > commit which partitions changed, regardless of the value from the watermark. > > Best regards > Theo > > ------------------------------ > *Von: *"Yun Gao" <yungao...@aliyun.com> > *An: *"Robert Metzger" <rmetz...@apache.org>, "Jingsong Li" < > jingsongl...@gmail.com> > *CC: *"Peter Groesbeck" <peter.groesb...@gmail.com>, "user" < > user@flink.apache.org> > *Gesendet: *Dienstag, 12. Mai 2020 10:36:59 > *Betreff: *回复：Re: Writing _SUCCESS Files (Streaming and Batch) > > Hi Peter, > > Sorry for missing the question and response later, I'm currently > sworking together with Jingsong on the issue to support "global committing" > (like writing _SUCCESS file or adding partitions to hive store) after > buckets terminated. In 1.11 we may first support watermark/time related > buckets in Table/SQL API, and we are also thinking of supporting "global > committing" for arbitrary bucket assigner policy for StreamingFileSink > users. The current rough thought is to let users specify when a bucket is > terminated on a single task, and the OperatorCoordinator[1] of the sink > will aggreate the information from all subtasks about this bucket and do > the global committing if the bucket has been finished on all the subtasks, > but this is still under thinking and discussion. Any thoughts or > requirements on this issue are warmly welcome. > > Best, > Yun > > > [1] OperatorCoordinator is introduced in FLIP-27: > https://cwiki.apache.org/confluence/display/FLINK/FLIP-27%3A+Refactor+Source+Interface. > This is a component resides in JobManager and could communicate with all > the subtasks of the corresponding operator, thus it could be used to > aggregate status from subtasks. > > ------------------原始邮件 ------------------ > *发件人:*Robert Metzger <rmetz...@apache.org> > *发送时间:*Tue May 12 15:36:26 2020 > *收件人:*Jingsong Li <jingsongl...@gmail.com> > *抄送:*Peter Groesbeck <peter.groesb...@gmail.com>, user < > user@flink.apache.org> > *主题:*Re: Writing _SUCCESS Files (Streaming and Batch) > >> Hi Peter, >> I filed a ticket for this feature request: >> https://issues.apache.org/jira/browse/FLINK-17627 (feel free to add your >> thoughts / requirements to the ticket) >> >> Best, >> Robert >> >> >> On Wed, May 6, 2020 at 3:41 AM Jingsong Li <jingsongl...@gmail.com> >> wrote: >> >>> Hi Peter, >>> The troublesome is how to know the "ending" for a bucket in streaming >>> job. >>> In 1.11, we are trying to implement a watermark-related bucket ending >>> mechanism[1] in Table/SQL. >>> >>> [1] >>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-115%3A+Filesystem+connector+in+Table >>> >>> Best, >>> Jingsong Lee >>> >>> On Tue, May 5, 2020 at 7:40 AM Peter Groesbeck < >>> peter.groesb...@gmail.com> wrote: >>> >>>> I am replacing an M/R job with a Streaming job using the >>>> StreamingFileSink and there is a requirement to generate an empty _SUCCESS >>>> file like the old Hadoop job. I have to implement a similar Batch job to >>>> read from backup files in case of outages or downtime. >>>> >>>> The Batch job question was answered here and appears to be still >>>> relevant although if someone could confirm for me that would be great. >>>> https://stackoverflow.com/a/39413810 >>>> >>>> The question of the Streaming job came up back in 2018 here: >>>> >>>> http://mail-archives.apache.org/mod_mbox/flink-user/201802.mbox/%3cff74eed5-602f-4eaa-9bc1-6cdf56611...@gmail.com%3E >>>> >>>> But the solution to use or extend the BucketingSink class seems out of >>>> date now that BucketingSink has been deprecated. >>>> >>>> Is there a way to implement a similar solution for StreamingFileSink? >>>> >>>> I'm currently on 1.8.1 although I hope to update to 1.10 in the near >>>> future. >>>> >>>> Thank you, >>>> Peter >>>> >>> >>> >>> -- >>> Best, Jingsong Lee >>> >>
