Hi all, @Yun Tang and I have an offline discussion, and we agreed that:
1. The design of this FLIP is pretty much like the option 3 in design doc[1] for FLINK-23342, and it is almost the best solution in general. Based on our production experience, this FLIP can solve the file flood problem very well. 2. There is a corner case that the directory may be left over when the job stops, so I added some content in section 4.8. Best, Zakelly [1] https://docs.google.com/document/d/1NJJQ30P27BmUvD7oa4FChvkYxMEgjRPTVdO1dHLl_9I/edit# On Fri, May 5, 2023 at 11:19 AM Zakelly Lan <zakelly....@gmail.com> wrote: > > Hi Rui Fan, > > Thanks for your reply. > > > Maybe re-uploaded in the next checkpoint is also a general solution > > for shared state? If yes, could we consider it as an optimization? > > And we can do it after the FLIP is done. > > Yes, it is a general solution for shared states. Maybe in the first > version we can let the shared states not re-use any previous state > handle after restoring, thus the state backend will do a full snapshot > and re-uploading the files it needs. This could cover the scenario > that rocksdb only uploads the base DB files. And later we could > consider performing fast copy in DFS to optimize the re-uploading. > WDYT? > > > > I'm not sure why we need 2 configurations, or whether 1 configuration > > is enough here. > > > The `max-file-pool-size` is hard to give a default value. For jobs with > > many tasks in a TM, it may be useful for file merging. However, > > it doesn't work well for jobs with a small number of tasks in a TM. > > > I prefer just adding the `max-file-pool-size`, and > > the `pool size = number of tasks / max-file-pool-size`. WDYT? > > > Sorry for not explaining clearly. The value of pool size is calculated by: > > 1. pool size = number of tasks / max-subtasks-per-file > 2. if pool size > max-file-pool-size then pool size = max-file-pool-size > > The `max-subtasks-per-file` addresses the issue of sequential file > writing, while the `max-file-pool-size` acts as a safeguard to prevent > an excessively large file pool. WDYT? > > > Thanks again for your thoughts. > > Best, > Zakelly > > On Thu, May 4, 2023 at 3:52 PM Rui Fan <1996fan...@gmail.com> wrote: > > > > Hi Zakelly, > > > > Sorry for the late reply, I still have some minor questions. > > > > >> (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. > > > > Maybe re-uploaded in the next checkpoint is also a general solution > > for shared state? If yes, could we consider it as an optimization? > > And we can do it after the FLIP is done. > > > > >> (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. > > > > I'm not sure why we need 2 configurations, or whether 1 configuration > > is enough here. > > > > The `max-file-pool-size` is hard to give a default value. For jobs with > > many tasks in a TM, it may be useful for file merging. However, > > it doesn't work well for jobs with a small number of tasks in a TM. > > > > I prefer just adding the `max-file-pool-size`, and > > the `pool size = number of tasks / max-file-pool-size`. WDYT? > > > > Maybe I missed some information. Please correct me if I'm wrong, thanks. > > > > Best, > > Rui Fan > > > > On Fri, Apr 28, 2023 at 12:10 AM Zakelly Lan <zakelly....@gmail.com> wrote: > > > > > 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 > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >