Hi Biao, Apologies for the late jumping in. My only question is about SinkFunction, does this imply that we want to add support for this to the SinkFunction? If so, I would not be in favour of that since we would like to deprecate (I actually thought that was already the case) the SinkFunction in favour of SinkV2.
Besides that, I have no other comments. Best regards, Martijn On Wed, Jan 4, 2023 at 7:28 AM Jing Zhang <beyond1...@gmail.com> wrote: > Hi Biao, > > Thanks for explanation. > > +1 for the proposal. > > Best, > Jing Zhang > > Lijie Wang <wangdachui9...@gmail.com> 于2023年1月4日周三 12:11写道： > > > Hi Biao, > > > > Thanks for the explanation of how SinkV2 knows the right subtask > > attempt. I have no more questions, +1 for the proposal. > > > > Best, > > Lijie > > > > Biao Liu <mmyy1...@gmail.com> 于2022年12月28日周三 17:22写道： > > > > > Thanks for all your feedback! > > > > > > To @Yuxia, > > > > > > > What the sink expect to do to isolate data produced by speculative > > > > executions? IIUC, if the taks failover, it also generate a new > > attempt. > > > > Does it make difference in isolating data produced? > > > > > > > > > Yes there is something different from the task failover scenario. The > > > attempt number is more necessary for speculative execution than > failover. > > > Because there can be only one subtask instance running at the same time > > in > > > the failover scenario. > > > > > > Let's take FileSystemOutputFormat as an example. For the failover > > scenario, > > > the temporary directory to store produced data can be something like > > > "$root_dir/task-$taskNumber/". At the initialization phase, subtask > > deletes > > > and re-creates the temporary directory. > > > > > > However in the speculative execution scenario, it does not work because > > > there might be several subtasks running at the same time. These > subtasks > > > might delete, re-create and write the same temporary directory at the > > > same time. The correct temporary directory should be like > > > "$root_dir/task-$taskNumber/attempt-$attemptNumber". So it's necessary > to > > > expose the attempt number to the Sink implementation to do the data > > > isolation. > > > > > > > > > To @Lijie, > > > > > > > I have a question about this: does SinkV2 need to do the same thing? > > > > > > > > > Actually, yes. > > > > > > Should we/users do it in the committer? If yes, how does the commiter > > know > > > > which one is the right subtask attempt? > > > > > > > > > Yes, we/users should do it in the committer. > > > > > > In the current design, the Committer of Sink V2 should get the "which > one > > > is the right subtask attempt" information from the "committable data'' > > > produced by SinkWriter. Let's take the FileSink as example, the > > > "committable data" sent to the Committer contains the full path of the > > > files produced by SinkWriter. Users could also pass the attempt number > > > through "committable data" from SinkWriter to Committer. > > > > > > In the "Rejected Alternatives -> Introduce a way to clean leaked data > of > > > Sink V2" section of the FLIP document, we discussed some of the reasons > > > that we didn't provide the API like OutputFormat. > > > > > > To @Jing Zhang > > > > > > I have a question about this: Speculative execution of Committer will > be > > > > disabled. > > > > > > I agree with your point and I saw the similar requirements to disable > > > speculative > > > > execution for specified operators. > > > > > > However the requirement is not supported currently. I think there > > > should be some > > > > place to describe how to support it. > > > > > > > > > In this FLIP design, the speculative execution of Committer of Sink V2 > > will > > > be disabled by Flink. It's not an optional operation. Users can not > > change > > > it. > > > And as you said, "disable speculative execution for specified > operators" > > is > > > not supported in the FLIP. Because it's a bit out of scope: "Sink > > Supports > > > Speculative Execution For Batch Job". I think it's better to start > > another > > > FLIP to discuss it. "Fine-grained control of enabling speculative > > execution > > > for operators" can be the title of that FLIP. And we can discuss there > > how > > > to enable or disable speculative execution for specified operators > > > including Committer and pre/post-committer of Sink V2. > > > > > > What do you think? > > > > > > Thanks, > > > Biao /'bɪ.aʊ/ > > > > > > > > > > > > On Wed, 28 Dec 2022 at 11:30, Jing Zhang <beyond1...@gmail.com> wrote: > > > > > > > Hi Biao, > > > > > > > > Thanks for driving this FLIP. It's meaningful to support speculative > > > > execution > > > > of sinks is important. > > > > > > > > I have a question about this: Speculative execution of Committer will > > be > > > > disabled. > > > > > > > > I agree with your point and I saw the similar requirements to disable > > > > speculative execution for specified operators. > > > > > > > > However the requirement is not supported currently. I think there > > should > > > be > > > > some place to describe how to support it. > > > > > > > > Best, > > > > Jing Zhang > > > > > > > > Lijie Wang <wangdachui9...@gmail.com> 于2022年12月27日周二 18:51写道： > > > > > > > > > Hi Biao, > > > > > > > > > > Thanks for driving this FLIP. > > > > > In this FLIP, it introduces "int getFinishedAttempt(int > > subtaskIndex)" > > > > for > > > > > OutputFormat to know which subtask attempt is the one marked as > > > finished > > > > by > > > > > JM and commit the right data. > > > > > I have a question about this: does SinkV2 need to do the same > thing? > > > > Should > > > > > we/users do it in the committer? If yes, how does the commiter know > > > which > > > > > one is the right subtask attempt? > > > > > > > > > > Best, > > > > > Lijie > > > > > > > > > > yuxia <luoyu...@alumni.sjtu.edu.cn> 于2022年12月27日周二 10:01写道： > > > > > > > > > > > HI, Biao. > > > > > > Thanks for driving this FLIP. > > > > > > After quick look of this FLIP, I have a question about "expose > the > > > > > attempt > > > > > > number which can be used to isolate data produced by speculative > > > > > executions > > > > > > with the same subtask id". > > > > > > What the sink expect to do to isolate data produced by > speculative > > > > > > executions? IIUC, if the taks failover, it also generate a new > > > > attempt. > > > > > > Does it make difference in isolating data produced? > > > > > > > > > > > > Best regards, > > > > > > Yuxia > > > > > > > > > > > > ----- 原始邮件 ----- > > > > > > 发件人: "Biao Liu" <mmyy1...@gmail.com> > > > > > > 收件人: "dev" <dev@flink.apache.org> > > > > > > 发送时间: 星期四, 2022年 12 月 22日 下午 8:16:21 > > > > > > 主题: [DISCUSS] FLIP-281 Sink Supports Speculative Execution For > > Batch > > > > Job > > > > > > > > > > > > Hi everyone, > > > > > > > > > > > > I would like to start a discussion about making Sink support > > > > speculative > > > > > > execution for batch jobs. This proposal is a follow up of > > "FLIP-168: > > > > > > Speculative Execution For Batch Job"[1]. Speculative execution is > > > very > > > > > > meaningful for batch jobs. And it would be more complete after > > > > supporting > > > > > > speculative execution of Sink. Please find more details in the > FLIP > > > > > > document > > > > > > [2]. > > > > > > > > > > > > Looking forward to your feedback. > > > > > > [1] > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-168%3A+Speculative+Execution+for+Batch+Job > > > > > > [2] > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-281+Sink+Supports+Speculative+Execution+For+Batch+Job > > > > > > > > > > > > Thanks, > > > > > > Biao /'bɪ.aʊ/ > > > > > > > > > > > > > > > > > > > > >
