Hi all, Thanks for all the feedback so far.
The discussion has been going on for some time, and all the comments and suggestions are addressed. So I would like to start a vote on this FLIP, which begins a week later (May. 5th at 10:00 AM GMT). If you have any concerns, please don't hesitate to follow up on this discussion. Best regards, Zakelly On Fri, Apr 28, 2023 at 12:03 AM Zakelly Lan <zakelly....@gmail.com> wrote: > > Hi Yuan, > > Thanks for sharing your thoughts. Like you said, the code changes and > complexities are shaded in the newly introduced file management in TM, > while the old file management remains the same. It is safe for us to > take a small step towards decentralized file management in this way. I > put the POC branch here[1] so everyone can check the code change. > > Best regards, > Zakelly > > [1] https://github.com/Zakelly/flink/tree/flip306_poc > > On Thu, Apr 27, 2023 at 8:13 PM Yuan Mei <yuanmei.w...@gmail.com> wrote: > > > > Hey all, > > > > Thanks @Zakelly for driving this effort and thanks everyone for the warm > > discussion. Sorry for the late response. > > > > As I and Zakelly have already discussed and reviewed the design carefully > > when drafting this FLIP, I do not have additional inputs here. But I want > > to highlight several points that I've been quoted and explain why I think > > the current design is a reasonable and clean one. > > > > *Why this FLIP is proposed* > > File Flooding is a problem for Flink I've seen many people bring up > > throughout the years, especially for large clusters. Unfortunately, there > > are not yet accepted solutions for the most commonly used state backend > > like RocksDB. This FLIP was originally targeted to address merging > > SST(KeyedState) checkpoint files. > > > > While we are comparing different design choices, we found that different > > types of checkpoint files (OPState, Unaligned CP channel state, Changelog > > incremental state) share similar considerations, for example, file > > management, file merging granularity, and e.t.c. That's why we want to > > abstract a unified framework for merging these different types of > > checkpoint files and provide flexibility to choose between merging > > efficiency, rescaling/restoring cost, File system capabilities (affecting > > File visibility), and e.t.c. > > > > *File Ownership moved from JM to TM, WHY* > > One of the major differences in the proposed design is moving file > > ownership from JM to TM. A lot of questions/concerns are coming from here, > > let me answer them one by one: > > > > *1. Why the current JM SharedRegistry is not enough and do we have to > > introduce more complexity?* > > SharedRegistry maintains the mapping between *a file -> max CP ID using the > > file* > > For merging files, we have to introduce another level of mapping *a file -> > > checkpoint file segment (merged files)* > > So yes, no matter what, the second level of mapping has to be managed > > somewhere, either JM or TM. > > > > *2. Why the **complexity (second level of mapping)** cannot be maintained > > in JM?* > > - As a centralized service, JM has already been complicated and overloaded. > > As mentioned by @Yanfei Lei <fredia...@gmail.com>, "triggering checkpoints > > can be delayed by discarding shared state when JM manages a large number of > > files FLINK-26590". This ends up setting the JM thread pool to 500! > > - As explained by @Zakelly in the previous thread, the contract "for > > Checkpoint N, only re-use shared state handles that have been already > > referenced by checkpoint N-1" is not guaranteed for the concurrent > > checkpoint in the current JM-owned design. This problem can not be > > addressed without significant changes in how SharedRegistry and checkpoint > > subsume work, which, I do not think is worth it since "concurrent_CP>1" is > > not used that much in prod. > > > > *3. We have similar discussions before, moving ownership from JM to TM, why > > it is not adopted at that time? * > > As mentioned by Yun and Piotr, we have had similar discussions to move > > ownership from JM to TM when designing the changelog state backend. The > > reason why we stuck to JM ownership at that time is mainly due to > > engineering time/effort constraints. > > This time, since we need an extra level of mapping, which complicates the > > JM logic even further, we indeed need to shade the complexity within the TM > > to avoid more communications between JM and TM. > > Zakelly has already shared the code branch (about 2000 lines), and it is > > simple. > > > > *4. Cloud-Native Trend* > > The current centralized file management framework contradicts the > > cloud-native trend. That's also one of the reasons moving ownership from JM > > to TM was first proposed. The proposed design and implementation is a > > worthy try-out in this direction. I'd like to put some more effort in this > > direction if this really turns out working well. > > > > One more thing I want to mention is that the proposed design shaded all the > > code changes and complexities in the newly introduced File management in > > TM. That says without enabling File merging, the code path of File managing > > remains the same as before. So it is also a safe change in this sense. > > > > Best, > > Yuan > > > > > > > > On Wed, Apr 12, 2023 at 5:23 PM Zakelly Lan <zakelly....@gmail.com> wrote: > > > > > Hi Yun, > > > > > > I reorganized our discussion and added a comparison table of state > > > ownership with some previous designs. Please take a look at section > > > "4.9. State ownership comparison with other designs". > > > > > > But I don't see them as alternatives since the design of state > > > ownership is integrated with this FLIP. That is to say, we are > > > providing a file merging solution including file management for new > > > merged files, other ownership models are not feasible for the current > > > merging plan. If the state ownership changes, the design of merging > > > files at different granularities also needs to be changed accordingly. > > > WDYT? > > > > > > > > > Best regards, > > > Zakelly > > > > > > On Tue, Apr 11, 2023 at 10:18 PM Yun Tang <myas...@live.com> wrote: > > > > > > > > Hi Zakelly, > > > > > > > > Since we already had some discussions on this topic in the doc I > > > mentioned, could you please describe the difference in your FLIP? > > > > > > > > I think we should better have a comparing table across different options > > > just like the doc wrote. And we could also list some of them in your > > > Rejected Alternatives part. > > > > > > > > > > > > Best > > > > Yun Tang > > > > ________________________________ > > > > From: Zakelly Lan <zakelly....@gmail.com> > > > > Sent: Tuesday, April 11, 2023 17:57 > > > > To: dev@flink.apache.org <dev@flink.apache.org> > > > > Subject: Re: [DISCUSS] FLIP-306: Unified File Merging Mechanism for > > > Checkpoints > > > > > > > > Hi Rui Fan, > > > > > > > > Thanks for your comments! > > > > > > > > > (1) The temporary segment will remain in the physical file for a short > > > time, right? > > > > > > > > Yes, any written segment will remain in the physical file until the > > > > physical file is deleted. It is controlled by the reference counting. > > > > And as discussed in 4.7, this will result in a space amplification > > > > problem. > > > > > > > > > > > > > (2) Is subtask granularity confused with shared state? > > > > > > > > Merging files at granularity of subtask is a general solution for > > > > shared states, considering the file may be reused by the following > > > > checkpoint after job restore. This design is applicable to sst files > > > > and any other shared states that may arise in the future. However, the > > > > DSTL files are a special case of shared states, since these files will > > > > no longer be shared after job restore. Therefore, we may do an > > > > optimization for these files and merge them at the TM level. > > > > Currently, the DSTL files are not in the shared directory of > > > > checkpoint storage, and I suggest we keep it as it is. I agree that > > > > this may bring in some confusion, and I suggest the FLIP mainly > > > > discuss the general situation and list the special situations > > > > separately without bringing in new concepts. I will add another > > > > paragraph describing the file merging for DSTL files. WDYT? > > > > > > > > > > > > > (3) When rescaling, do all shared files need to be copied? > > > > > > > > I agree with you that only sst files of the base DB need to be copied > > > > (or re-uploaded in the next checkpoint). However, section 4.2 > > > > simplifies file copying issues (copying all files), following the > > > > concept of shared state. > > > > > > > > > > > > > (4) Does the space magnification ratio need a configuration option? > > > > > > > > Thanks for the reminder, I will add an option in this FLIP. > > > > > > > > > > > > > (5) How many physical files can a TM write at the same checkpoint at > > > the same time? > > > > > > > > This is a very good point. Actually, there is a file reuse pool as > > > > section 4.6 described. There could be multiple files within this pool, > > > > supporting concurrent writing by multiple writers. I suggest providing > > > > two configurations to control the file number: > > > > > > > > state.checkpoints.file-merging.max-file-pool-size: Specifies the > > > > upper limit of the file pool size. > > > > state.checkpoints.file-merging.max-subtasks-per-file: Specifies the > > > > lower limit of the file pool size based on the number of subtasks > > > > within each TM. > > > > > > > > The number of simultaneously open files is controlled by these two > > > > options, and the first option takes precedence over the second. > > > > > > > > WDYT? > > > > > > > > > > > > > > > > Thanks a lot for your valuable insight. > > > > > > > > Best regards, > > > > Zakelly > > > > > > > > > > > > On Mon, Apr 10, 2023 at 7:08 PM Rui Fan <1996fan...@gmail.com> wrote: > > > > > > > > > > Hi all, > > > > > > > > > > Thanks Zakelly driving this proposal, and thank you all for > > > > > the warm discussions. It's really a useful feature. > > > > > > > > > > I have a few questions about this FLIP. > > > > > > > > > > (1) The temporary segment will remain in the physical file for > > > > > a short time, right? > > > > > > > > > > FLIP proposes to write segments instead of physical files. > > > > > If the physical files are written directly, these temporary files > > > > > will be deleted after the checkpoint is aborted. When writing > > > > > a segment, how to delete the temporary segment? > > > > > Decrement the reference count value by 1? > > > > > > > > > > (2) Is subtask granularity confused with shared state? > > > > > > > > > > From the "4.1.2 Merge files within a subtask or a TM" part, > > > > > based on the principle of sst files, it is concluded that > > > > > "For shared states, files are merged within each subtask." > > > > > > > > > > I'm not sure whether this conclusion is general or just for sst. > > > > > As Yanfei mentioned before: > > > > > > > > > > > DSTL files are shared between checkpoints, and are > > > > > > currently merged in batches at the task manager level. > > > > > > > > > > DSTL files as the shared state in FLIP-306, however, it > > > > > would be better to merge at TM granularity. So, I'm not > > > > > sure whether the subtask granularity confused with > > > > > shared state? > > > > > > > > > > And I'm not familiar with DSTL file merging, should > > > > > shared state be divided into shared subtask state > > > > > and shared TM state? > > > > > > > > > > (3) When rescaling, do all shared files need to be copied? > > > > > > > > > > From the "4.2 Rescaling and Physical File Lifecycle" part, > > > > > I see a lot of file copying. > > > > > > > > > > As I understand, only sst files of the baseDB need to be copied. > > > > > From the restore code[1], when restoreWithRescaling, flink will > > > > > init a base DB instance, read all contents from other temporary > > > > > rocksdb instances, and write them into the base DB, and then > > > > > the temporary rocksdb instance will be discarded. > > > > > > > > > > So, I think copying the files of the base rocksdb is enough, and > > > > > the files of other rocksdb instances aren't used. > > > > > > > > > > Or do not copy any files during recovery, upload all sst files > > > > > at the first checkpoint. > > > > > > > > > > (4) Does the space magnification ratio need a configuration option? > > > > > > > > > > From the step1 of "4.7 Space amplification" part, I see: > > > > > > > > > > > Checking whether the space amplification of each file is greater > > > than a > > > > > preset threshold and collecting files that exceed the threshold for > > > > > compaction. > > > > > > > > > > Should we add a configuration option about the compaction threshold? > > > > > I didn't see it at "5. Public interfaces and User Cases" part. > > > > > > > > > > (5) How many physical files can a TM write at the same > > > > > checkpoint at the same time? > > > > > > > > > > From the "5. Public interfaces and User Cases" part, I see: > > > > > > > > > > > A configuration option that sets a maximum size limit for physical > > > files. > > > > > > > > > > I guess that each type of state(private or shared state) will only > > > > > write one file at the same time at the same checkpoint. > > > > > When the file reaches the maximum size, flink will start writing > > > > > the next file, right? > > > > > > > > > > If yes, for shared state, will > > > > > "state.backend.rocksdb.checkpoint.transfer.thread.num" > > > > > be invalid? > > > > > > > > > > For private state, a TM may have many tasks (because of slot > > > > > sharing, more than 20 tasks may run in a slot), and the > > > > > performance of all tasks serially writing files may be poor, > > > > > eventually resulting in longer checkpoint time. > > > > > > > > > > That's why FLINK-26803[2] introduced a configuration option: > > > > > > > > "execution.checkpointing.unaligned.max-subtasks-per-channel-state-file". > > > > > Flink users can set the maximum number of subtasks that > > > > > share the same channel state file. > > > > > > > > > > That's all my questions right now, please correct me if > > > > > anything is wrong. > > > > > > > > > > Anyway, this FLIP is useful for the stability of large-scale > > > > > flink production. Looking forward to its completion and > > > > > eventual acceptance by the community. > > > > > > > > > > [1] > > > > > > > > https://github.com/apache/flink/blob/65710b437318364ec19c0369d038ac2222c10498/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/restore/RocksDBIncrementalRestoreOperation.java#L292 > > > > > [2] https://issues.apache.org/jira/browse/FLINK-26803 > > > > > > > > > > Best, > > > > > Rui Fan > > > > > > > > > > On Fri, Apr 7, 2023 at 8:42 PM Jing Ge <j...@ververica.com.invalid> > > > wrote: > > > > > > > > > > > Hi, > > > > > > > > > > > > Jingsong, Yanfei, please check, if you can view the doc. Thanks. > > > > > > > > > > > > Best regards, > > > > > > Jing > > > > > > > > > > > > On Fri, Apr 7, 2023 at 2:19 PM Zakelly Lan <zakelly....@gmail.com> > > > wrote: > > > > > > > > > > > > > Hi Yanfei, > > > > > > > > > > > > > > Thanks for your comments. > > > > > > > > > > > > > > > Does this result in a larger space amplification? Maybe a more > > > > > > > suitable value can be determined through some experimental > > > statistics > > > > > > > after we implement this feature. > > > > > > > > > > > > > > Yes, it results in larger space amplification for shared states. I > > > > > > > will do more tests and investigation. > > > > > > > > > > > > > > > > > > > > > Thanks. > > > > > > > > > > > > > > Best regards, > > > > > > > Zakelly > > > > > > > > > > > > > > On Fri, Apr 7, 2023 at 8:15 PM Zakelly Lan <zakelly....@gmail.com> > > > > > > wrote: > > > > > > > > > > > > > > > > Hi @Piotr and @Jingsong Li > > > > > > > > > > > > > > > > I have read access to the document, but I'm not sure whether the > > > owner > > > > > > > > of this document wants to make it public. Actually, the doc is > > > for > > > > > > > > FLINK-23342 and there is a candidate design very similar to this > > > FLIP, > > > > > > > > but only for the shared state. Like Yun said, the previous > > > design is > > > > > > > > not taken because of the code complexity, however I think it is > > > > > > > > acceptable after implementing the POC[1]. I think we could focus > > > on > > > > > > > > the current plan, WDTY? > > > > > > > > > > > > > > > > > > > > > > > > [1] POC of this FLIP: > > > > > > > > > > > > > > > > > > > > > > > > https://github.com/Zakelly/flink/commit/98538185182e33739828ee36ab96dcf2aebad80c > > > > > > > > > > > > > > > > On Fri, Apr 7, 2023 at 8:13 PM Zakelly Lan < > > > zakelly....@gmail.com> > > > > > > > wrote: > > > > > > > > > > > > > > > > > > Hi Piotr, > > > > > > > > > > > > > > > > > > Thanks for your comments! > > > > > > > > > > > > > > > > > > (1) Sorry for the misleading, let me make it more clear. It is > > > a > > > > > > > > > concurrent checkpoint senario. Yes, the assumption you said > > > needs to > > > > > > > > > be followed, but the state handles here refer to the original > > > SST > > > > > > > > > files, not the underlying file. In this FLIP when checkpoint N > > > and > > > > > > N+1 > > > > > > > > > are running concurrently, they reuse files from checkpoint > > > N-1, and > > > > > > > > > some of the files may be deleted when checkpoint N completes > > > while > > > > > > > > > checkpoint N+1 is still writing on it. There is no such > > > problem for > > > > > > > > > original shared states without file merging because when a > > > state > > > > > > > > > handle (or sst file here) from checkpoint N-1 is not > > > referenced by > > > > > > > > > checkpoint N, it will not be referenced by checkpoint N+1. So > > > the > > > > > > > > > subsumption of sst files from checkpoint N-1 are safe. > > > > > > > > > For above example, when reaching step "d.", File 1 reached the > > > size > > > > > > > > > threshold and will not be used. The Chk-2 and Chk-3 are > > > > > > > > > running > > > > > > > > > concurrently, and the File 3 is being written by Chk-2, so it > > > can not > > > > > > > > > be used by Chk-3 (As described in section 4.6). Here comes the > > > > > > > > > problem. > > > > > > > > > > > > > > > > > > (2) Please correct me if I'm wrong. The purpose of the > > > > > > > > > `RecoverableWriter` is to provide a reliable file writer even > > > > > > > > > tolerable with job failure and recovery. The implementation > > > varies > > > > > > > > > among the file systems, some of which involves writing into > > > temporary > > > > > > > > > files (such as HDFS). As a result, it may produce more RPC > > > requests > > > > > > to > > > > > > > > > the DFS. > > > > > > > > > The goal of this FLIP is to reduce the pressure on DFS, > > > especially > > > > > > the > > > > > > > > > number of files and RPC requests. Currently the TMs are NOT > > > using the > > > > > > > > > RecoverableWriter to persist/upload the state files, and a > > > > > > > > > file > > > > > > > > > closing is enough. The section 4.1.1 is trying to omit this > > > file > > > > > > > > > closing but ensure file visibility in some DFS, thus reducing > > > > > > pressure > > > > > > > > > on DFS. That's why I said the problems they want to solve are > > > > > > > > > different. I'm not sure if I made myself clear. > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks! > > > > > > > > > > > > > > > > > > Best regards, > > > > > > > > > Zakelly > > > > > > > > > > > > > > > > > > On Fri, Apr 7, 2023 at 8:08 PM Zakelly Lan < > > > zakelly....@gmail.com> > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > Hi Yun, > > > > > > > > > > > > > > > > > > > > Thanks for your suggestions! > > > > > > > > > > > > > > > > > > > > I have read the FLINK-23342 and its design doc as you > > > provided. > > > > > > First > > > > > > > > > > of all the goal of this FLIP and the doc are similar, and > > > > > > > > > > the > > > > > > design > > > > > > > > > > of this FLIP is pretty much like option 3. The main > > > difference is > > > > > > > that > > > > > > > > > > we imply the concept of 'epoch' in the folder path for each > > > > > > > > > > granularity. For shared state, the folder for each subtask > > > is like > > > > > > > > > > "${checkpointBaseDir}/shared/subtask-{index}-{parallelism}", > > > so if > > > > > > > the > > > > > > > > > > ${checkpointBaseDir} changes (when user restart a job > > > manually) or > > > > > > > the > > > > > > > > > > ${parallelism} changes (when rescaling), there will be a > > > > > > > re-uploading, > > > > > > > > > > and the JM takes care of the old artifacts. The folder path > > > for > > > > > > > > > > private state is in the form of > > > > > > > > > > "${checkpointBaseDir}/tm-owned/${tmResourceId}" and the > > > division of > > > > > > > > > > responsibilities between JM and TM is similar. The design of > > > this > > > > > > > FLIP > > > > > > > > > > inherits all the advantages of the design of option 3 in > > > that doc, > > > > > > > and > > > > > > > > > > also avoids extra communication for epoch maintenance. As > > > for the > > > > > > > code > > > > > > > > > > complexity, you may check the POC commit[1] and find that > > > > > > > > > > the > > > > > > > > > > implementation is pretty clean and is a totally new code > > > > > > > > > > path > > > > > > making > > > > > > > > > > nearly no influence on the old one. Comparing the number of > > > lines > > > > > > of > > > > > > > > > > code change with what's currently done for merging channel > > > state[2] > > > > > > > > > > (5200 vs. 2500 additions), I think it is acceptable > > > considering we > > > > > > > are > > > > > > > > > > providing a unified file merging framework, which would save > > > a lot > > > > > > of > > > > > > > > > > effort in future. WDYT? > > > > > > > > > > > > > > > > > > > > Thanks. > > > > > > > > > > > > > > > > > > > > Best regards, > > > > > > > > > > Zakelly > > > > > > > > > > > > > > > > > > > > [1] POC of this FLIP: > > > > > > > > > > > > > > > > > > > > > > > > > > https://github.com/Zakelly/flink/commit/98538185182e33739828ee36ab96dcf2aebad80c > > > > > > > > > > [2] Commit for FLINK-26803 (Merge the channel state files) : > > > > > > > > > > > > > > > > > > > > > > > > > > https://github.com/apache/flink/commit/8be94e6663d8ac6e3d74bf4cd5f540cc96c8289e > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Fri, Apr 7, 2023 at 7:22 PM Yanfei Lei < > > > fredia...@gmail.com> > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > Thanks for your explanation Zakelly. > > > > > > > > > > > (1) Keeping these merging granularities for different > > > types of > > > > > > > files > > > > > > > > > > > as presets that are not configurable is a good idea to > > > prevent > > > > > > > > > > > performance degradation. > > > > > > > > > > > > > > > > > > > > > > (2) > > > > > > > > > > > > For the third option, 64MB is an acceptable target size. > > > The > > > > > > > RocksDB state backend in Flink also chooses 64MB as the default > > > target > > > > > > file > > > > > > > size. > > > > > > > > > > > > > > > > > > > > > > Does this result in a larger space amplification? Maybe a > > > more > > > > > > > > > > > suitable value can be determined through some experimental > > > > > > > statistics > > > > > > > > > > > after we implement this feature. > > > > > > > > > > > > > > > > > > > > > > Best, > > > > > > > > > > > Yanfei > > > > > > > > > > > > > > > > > > > > > > Jingsong Li <jingsongl...@gmail.com> 于2023年4月7日周五 17:09写道： > > > > > > > > > > > > > > > > > > > > > > > > Hi Yun, > > > > > > > > > > > > > > > > > > > > > > > > It looks like this doc needs permission to read? [1] > > > > > > > > > > > > > > > > > > > > > > > > [1] > > > > > > > > > > > > > > > > https://docs.google.com/document/d/1NJJQ30P27BmUvD7oa4FChvkYxMEgjRPTVdO1dHLl_9I/edit# > > > > > > > > > > > > > > > > > > > > > > > > Best, > > > > > > > > > > > > Jingsong > > > > > > > > > > > > > > > > > > > > > > > > On Fri, Apr 7, 2023 at 4:34 PM Piotr Nowojski < > > > > > > > pnowoj...@apache.org> wrote: > > > > > > > > > > > > > > > > > > > > > > > > > > Hi, > > > > > > > > > > > > > > > > > > > > > > > > > > +1 To what Yun Tang wrote. We don't seem to have > > > access to > > > > > > the > > > > > > > design doc. > > > > > > > > > > > > > Could you make it publicly visible or copy out its > > > content to > > > > > > > another > > > > > > > > > > > > > document? > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks for your answers Zakelly. > > > > > > > > > > > > > > > > > > > > > > > > > > (1) > > > > > > > > > > > > > Yes, the current mechanism introduced in FLINK-24611 > > > allows > > > > > > > for checkpoint > > > > > > > > > > > > > N, to only re-use shared state handles that have been > > > already > > > > > > > referenced by > > > > > > > > > > > > > checkpoint N-1. But why do we need to break this > > > assumption? > > > > > > > In your step, > > > > > > > > > > > > > "d.", TM could adhere to that assumption, and instead > > > of > > > > > > > reusing File-2, it > > > > > > > > > > > > > could either re-use File-1, File-3 or create a new > > > file. > > > > > > > > > > > > > > > > > > > > > > > > > > (2) > > > > > > > > > > > > > Can you elaborate a bit more on this? As far as I > > > recall, the > > > > > > > purpose of > > > > > > > > > > > > > the `RecoverableWriter` is to support exactly the > > > things > > > > > > > described in this > > > > > > > > > > > > > FLIP, so what's the difference? If you are saying that > > > for > > > > > > > this FLIP you > > > > > > > > > > > > > can implement something more efficiently for a given > > > > > > > FileSystem, then why > > > > > > > > > > > > > can it not be done the same way for the > > > `RecoverableWriter`? > > > > > > > > > > > > > > > > > > > > > > > > > > Best, > > > > > > > > > > > > > Piotrek > > > > > > > > > > > > > > > > > > > > > > > > > > czw., 6 kwi 2023 o 17:24 Yun Tang <myas...@live.com> > > > > > > > napisał(a): > > > > > > > > > > > > > > > > > > > > > > > > > > > Hi Zakelly, > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks for driving this work! > > > > > > > > > > > > > > > > > > > > > > > > > > > > I'm not sure did you ever read the discussion > > > > > > > > > > > > > > between > > > > > > > Stephan, Roman, > > > > > > > > > > > > > > Piotr, Yuan and I in the design doc [1] in nearly > > > two years > > > > > > > ago. > > > > > > > > > > > > > > > > > > > > > > > > > > > > From my understanding, your proposal is also a mixed > > > state > > > > > > > ownership: some > > > > > > > > > > > > > > states are owned by the TM while some are owned by > > > the JM. > > > > > > > If my memory is > > > > > > > > > > > > > > correct, we did not take the option-3 or option-5 in > > > the > > > > > > > design doc [1] for > > > > > > > > > > > > > > the code complexity when implements the 1st version > > > of > > > > > > > changelog > > > > > > > > > > > > > > state-backend. > > > > > > > > > > > > > > > > > > > > > > > > > > > > Could you also compare the current FLIP with the > > > proposals > > > > > > > in the design > > > > > > > > > > > > > > doc[1]? From my understanding, we should at least > > > consider > > > > > > > to comapre with > > > > > > > > > > > > > > option-3 and option-5 as they are all mixed > > > solutions. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > [1] > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://docs.google.com/document/d/1NJJQ30P27BmUvD7oa4FChvkYxMEgjRPTVdO1dHLl_9I/edit# > > > > > > > > > > > > > > > > > > > > > > > > > > > > Best > > > > > > > > > > > > > > Yun Tang > > > > > > > > > > > > > > > > > > > > > > > > > > > > ------------------------------ > > > > > > > > > > > > > > *From:* Zakelly Lan <zakelly....@gmail.com> > > > > > > > > > > > > > > *Sent:* Thursday, April 6, 2023 16:38 > > > > > > > > > > > > > > *To:* dev@flink.apache.org <dev@flink.apache.org> > > > > > > > > > > > > > > *Subject:* Re: [DISCUSS] FLIP-306: Unified File > > > Merging > > > > > > > Mechanism for > > > > > > > > > > > > > > Checkpoints > > > > > > > > > > > > > > > > > > > > > > > > > > > > Hi Piotr, > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks for all the feedback. > > > > > > > > > > > > > > > > > > > > > > > > > > > > (1) Thanks for the reminder. I have just seen the > > > > > > > FLINK-24611, the delayed > > > > > > > > > > > > > > deletion by JM resolves some sync problems between > > > JM and > > > > > > > TM, but I'm > > > > > > > > > > > > > > afraid it is still not feasible for the file sharing > > > in > > > > > > this > > > > > > > FLIP. > > > > > > > > > > > > > > Considering a concurrent checkpoint scenario as > > > follows: > > > > > > > > > > > > > > a. Checkpoint 1 finishes. 1.sst, 2.sst and 3.sst > > > are > > > > > > > written in file 1, > > > > > > > > > > > > > > and 4.sst is written in file 2. > > > > > > > > > > > > > > b. Checkpoint 2 starts based on checkpoint 1, > > > including > > > > > > > 1.sst, 2.sst > > > > > > > > > > > > > > and 5.sst. > > > > > > > > > > > > > > c. Checkpoint 3 starts based on checkpoint 1, > > > including > > > > > > > 1.sst, 2.sst > > > > > > > > > > > > > > and 5.sst as well. > > > > > > > > > > > > > > d. Checkpoint 3 reuses the file 2, TM writes > > > 5.sst on > > > > > > it. > > > > > > > > > > > > > > e. Checkpoint 2 creates a new file 3, TM writes > > > 5.sst on > > > > > > > it. > > > > > > > > > > > > > > f. Checkpoint 2 finishes, checkpoint 1 is > > > subsumed and > > > > > > > the file 2 is > > > > > > > > > > > > > > deleted, while checkpoint 3 still needs file 2. > > > > > > > > > > > > > > > > > > > > > > > > > > > > I attached a diagram to describe the scenario. > > > > > > > > > > > > > > [image: concurrent cp.jpg] > > > > > > > > > > > > > > The core issue is that this FLIP introduces a > > > mechanism > > > > > > that > > > > > > > allows > > > > > > > > > > > > > > physical files to be potentially used by the next > > > several > > > > > > > checkpoints. JM > > > > > > > > > > > > > > is uncertain whether there will be a TM continuing > > > to write > > > > > > > to a specific > > > > > > > > > > > > > > file. So in this FLIP, TMs take the responsibility > > > > > > > > > > > > > > to > > > > > > delete > > > > > > > the physical > > > > > > > > > > > > > > files. > > > > > > > > > > > > > > > > > > > > > > > > > > > > (2) IIUC, the RecoverableWriter is introduced to > > > persist > > > > > > > data in the "in > > > > > > > > > > > > > > progress" files after each checkpoint, and the > > > > > > > implementation may be based > > > > > > > > > > > > > > on the file sync in some file systems. However, > > > since the > > > > > > > sync is a heavy > > > > > > > > > > > > > > operation for DFS, this FLIP wants to use flush > > > instead of > > > > > > > the sync with > > > > > > > > > > > > > > the best effort. This only fits the case that the > > > DFS is > > > > > > > considered > > > > > > > > > > > > > > reliable. The problems they want to solve are > > > different. > > > > > > > > > > > > > > > > > > > > > > > > > > > > (3) Yes, if files are managed by JM via the shared > > > > > > registry, > > > > > > > this problem > > > > > > > > > > > > > > is solved. And as I mentioned in (1), there are some > > > other > > > > > > > corner cases > > > > > > > > > > > > > > hard to resolve via the shared registry. > > > > > > > > > > > > > > > > > > > > > > > > > > > > The goal of this FLIP is to have a common way of > > > merging > > > > > > > files in all use > > > > > > > > > > > > > > cases. For shared state it merges at subtask level, > > > while > > > > > > > for private state > > > > > > > > > > > > > > (and changelog files, as I replied to Yanfei), files > > > are > > > > > > > merged at TM > > > > > > > > > > > > > > level. So it is not contrary to the current plan for > > > the > > > > > > > unaligned > > > > > > > > > > > > > > checkpoint state (FLINK-26803). You are right that > > > the > > > > > > > unaligned checkpoint > > > > > > > > > > > > > > state would be merged with the operator's state > > > file, so > > > > > > > overall, it is > > > > > > > > > > > > > > slightly better than what's currently done. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks again for the valuable comments! > > > > > > > > > > > > > > > > > > > > > > > > > > > > Best regards, > > > > > > > > > > > > > > Zakelly > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Wed, Apr 5, 2023 at 8:43 PM Piotr Nowojski < > > > > > > > pnowoj...@apache.org> > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > > > > Hi, > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks for coming up with the proposal, it's > > > definitely > > > > > > > valuable. I'm still > > > > > > > > > > > > > > reading and trying to understand the proposal, but a > > > couple > > > > > > > of comments > > > > > > > > > > > > > > from my side. > > > > > > > > > > > > > > > > > > > > > > > > > > > > (1) > > > > > > > > > > > > > > > Ownership of a single checkpoint file is > > > transferred to > > > > > > > TM, while JM > > > > > > > > > > > > > > manages the parent directory of these files. > > > > > > > > > > > > > > > > > > > > > > > > > > > > Have you seen > > > > > > > https://issues.apache.org/jira/browse/FLINK-24611 before? I > > > > > > > > > > > > > > don't fully remember why, but we have rejected the > > > idea of > > > > > > > moving the file > > > > > > > > > > > > > > ownership to TM and instead reworked the shared file > > > > > > > registry in a way that > > > > > > > > > > > > > > I think should be sufficient for file sharing. Could > > > you > > > > > > > elaborate why we > > > > > > > > > > > > > > need to move the file ownership to TM, and why is > > > > > > > > > > > > > > the > > > > > > > current mechanism not > > > > > > > > > > > > > > sufficient? > > > > > > > > > > > > > > > > > > > > > > > > > > > > (2) > > > > > > > > > > > > > > > File visibility is needed when a Flink job > > > recovers after > > > > > > > a checkpoint is > > > > > > > > > > > > > > materialized. In some DFS, such as most object > > > storages, a > > > > > > > file is only > > > > > > > > > > > > > > visible after it is closed > > > > > > > > > > > > > > > > > > > > > > > > > > > > Is that really the case? > > > > > > > > > > > > > > > > > > > > > `org.apache.flink.core.fs.FileSystem#createRecoverableWriter` > > > seems to be > > > > > > > > > > > > > > addressing exactly this issue, and the most > > > frequently used > > > > > > > FileSystem (S3) > > > > > > > > > > > > > > AFAIK supports it with no problems? > > > > > > > > > > > > > > > > > > > > > > > > > > > > (3) > > > > > > > > > > > > > > > 4.1.2 Merge files within a subtask or a TM > > > > > > > > > > > > > > > Given that TMs are reassigned after restoration, > > > it is > > > > > > > difficult to > > > > > > > > > > > > > > manage physical files that contain data from > > > > > > > > > > > > > > multiple > > > > > > > subtasks scattered > > > > > > > > > > > > > > across different TMs (as depicted in Fig.3). There > > > is no > > > > > > > synchronization > > > > > > > > > > > > > > mechanism between TMs, making file management in > > > > > > > > > > > > > > this > > > > > > > scenario challenging. > > > > > > > > > > > > > > > > > > > > > > > > > > > > I think this is solved in many places already via > > > > > > > > > > > > > > the > > > > > > shared > > > > > > > state managed > > > > > > > > > > > > > > by the JM, as I mentioned in (1). > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > If I understand it correctly you are proposing to > > > have a > > > > > > > common > > > > > > > > > > > > > > interface/way of merging small files, in all use > > > cases, > > > > > > that > > > > > > > would work > > > > > > > > > > > > > > only across a single subtask? That's contrary to > > > what's > > > > > > > currently done for > > > > > > > > > > > > > > unaligned checkpoints, right? But if this generic > > > mechanism > > > > > > > was to be used > > > > > > > > > > > > > > for unaligned checkpoints, unaligned checkpoint > > > state would > > > > > > > have been > > > > > > > > > > > > > > merged with the operators state file, so all in all > > > there > > > > > > > would be no > > > > > > > > > > > > > > regression visible to a user? The limit is that we > > > always > > > > > > > have at least a > > > > > > > > > > > > > > single file per subtask, but in exchange we are > > > getting a > > > > > > > simpler threading > > > > > > > > > > > > > > model? > > > > > > > > > > > > > > > > > > > > > > > > > > > > Bets, > > > > > > > > > > > > > > Piotrek > > > > > > > > > > > > > > > > > > > > > > > > > > > > wt., 4 kwi 2023 o 08:51 Zakelly Lan < > > > zakelly....@gmail.com > > > > > > > > > > > > > > napisał(a): > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Hi Yanfei, > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thank you for your prompt response. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > I agree that managing (deleting) only some folders > > > with > > > > > > JM > > > > > > > can greatly > > > > > > > > > > > > > > > relieve JM's burden. Thanks for pointing this out. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > In general, merging at the TM level is more > > > effective > > > > > > > since there are > > > > > > > > > > > > > > > usually more files to merge. Therefore, I believe > > > it is > > > > > > > better to > > > > > > > > > > > > > > > merge files per TM as much as possible. However, > > > for > > > > > > > shared state, > > > > > > > > > > > > > > > merging at the subtask level is the best choice to > > > > > > prevent > > > > > > > significant > > > > > > > > > > > > > > > data transfer over the network after restoring. I > > > think > > > > > > it > > > > > > > is better > > > > > > > > > > > > > > > to keep these merging granularities for different > > > types > > > > > > of > > > > > > > files as > > > > > > > > > > > > > > > presets that are not configurable. WDYT? > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > As for the DSTL files, they are merged per TM and > > > placed > > > > > > > in the > > > > > > > > > > > > > > > task-owned folder. These files can be classified > > > > > > > > > > > > > > > as > > > > > > shared > > > > > > > state since > > > > > > > > > > > > > > > they are shared across checkpoints. However, the > > > DSTL > > > > > > file > > > > > > > is a > > > > > > > > > > > > > > > special case that will be subsumed by the first > > > > > > checkpoint > > > > > > > of the > > > > > > > > > > > > > > > newly restored job. Therefore, there is no need > > > for new > > > > > > > TMs to keep > > > > > > > > > > > > > > > these files after the old checkpoint is subsumed, > > > just > > > > > > > like the > > > > > > > > > > > > > > > private state files. Thus, it is feasible to merge > > > DSTL > > > > > > > files per TM > > > > > > > > > > > > > > > without introducing complex file management across > > > job > > > > > > > attempts. So > > > > > > > > > > > > > > > the possible performance degradation is avoided. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > The three newly introduced options have > > > > > > > > > > > > > > > recommended > > > > > > > defaults. For > > > > > > > > > > > > > > > upcoming versions, this feature is turned off by > > > default. > > > > > > > For the > > > > > > > > > > > > > > > second option, SEGMENTED_ACROSS_CP_BOUNDARY is the > > > > > > > recommended default > > > > > > > > > > > > > > > as it is more effective. Of course, if > > > encountering some > > > > > > > DFS that does > > > > > > > > > > > > > > > not support file visibility until the file is > > > closed, it > > > > > > > is possible > > > > > > > > > > > > > > > to fall back to another option automatically. For > > > the > > > > > > > third option, > > > > > > > > > > > > > > > 64MB is an acceptable target size. The RocksDB > > > state > > > > > > > backend in Flink > > > > > > > > > > > > > > > also chooses 64MB as the default target file size. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thank you again for your quick response. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Best regards, > > > > > > > > > > > > > > > Zakelly > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Mon, Apr 3, 2023 at 11:27 PM Yanfei Lei < > > > > > > > fredia...@gmail.com> wrote: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Hi Zakelly, > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks for driving this, this proposal enables > > > the > > > > > > > files merging of > > > > > > > > > > > > > > > > different types of states to be grouped under a > > > unified > > > > > > > framework. I > > > > > > > > > > > > > > > > think it has the added benefit of lightening the > > > load > > > > > > on > > > > > > > JM. As > > > > > > > > > > > > > > > > FLINK-26590[1] described, triggered checkpoints > > > can be > > > > > > > delayed by > > > > > > > > > > > > > > > > discarding shared state when JM manages a large > > > number > > > > > > > of files. After > > > > > > > > > > > > > > > > this FLIP, JM only needs to manage some folders, > > > which > > > > > > > greatly reduces > > > > > > > > > > > > > > > > the burden on JM. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > In Section 4.1, two types of merging > > > granularities(per > > > > > > > subtask and per > > > > > > > > > > > > > > > > task manager) are proposed, the shared state is > > > managed > > > > > > > by per subtask > > > > > > > > > > > > > > > > granularity, but for the changelog state > > > backend, its > > > > > > > DSTL files are > > > > > > > > > > > > > > > > shared between checkpoints, and are currently > > > merged in > > > > > > > batches at the > > > > > > > > > > > > > > > > task manager level. When merging with the > > > > > > > SEGMENTED_WITHIN_CP_BOUNDARY > > > > > > > > > > > > > > > > mode, I'm concerned about the performance > > > degradation > > > > > > of > > > > > > > its merging, > > > > > > > > > > > > > > > > hence I wonder if the merge granularities are > > > > > > > configurable? Further, > > > > > > > > > > > > > > > > from a user perspective, three new options are > > > > > > > introduced in this > > > > > > > > > > > > > > > > FLIP, do they have recommended defaults? > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > [1] > > > https://issues.apache.org/jira/browse/FLINK-26590 > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Best, > > > > > > > > > > > > > > > > Yanfei > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Zakelly Lan <zakelly....@gmail.com> 于2023年4月3日周一 > > > > > > > 18:36写道： > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Hi everyone, > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > I would like to open a discussion on providing > > > a > > > > > > > unified file merging > > > > > > > > > > > > > > > > > mechanism for checkpoints[1]. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Currently, many files are uploaded to the DFS > > > during > > > > > > > checkpoints, > > > > > > > > > > > > > > > > > leading to the 'file flood' problem when > > > running > > > > > > > > > > > > > > > > > intensive workloads in a cluster. To tackle > > > this > > > > > > > problem, various > > > > > > > > > > > > > > > > > solutions have been proposed for different > > > types > > > > > > > > > > > > > > > > > of state files. Although these methods are > > > similar, > > > > > > > they lack a > > > > > > > > > > > > > > > > > systematic view and approach. We believe that > > > it is > > > > > > > > > > > > > > > > > better to consider this problem as a whole and > > > > > > > introduce a unified > > > > > > > > > > > > > > > > > framework to address the file flood problem > > > > > > > > > > > > > > > > > for > > > > > > > > > > > > > > > > > all types of state files. A POC has been > > > implemented > > > > > > > based on current > > > > > > > > > > > > > > > > > FLIP design, and the test results are > > > promising. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Looking forward to your comments or feedback. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Best regards, > > > > > > > > > > > > > > > > > Zakelly > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > [1] > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-306%3A+Unified+File+Merging+Mechanism+for+Checkpoints > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >
