Hi Piotr. Thanks for your proposal. I have some comments, PTAL: 1. +1 about unifying the interface with DuplicatingFileSystem. IIUC, DuplicatingFileSystem also covers the logic from/to both local and remote paths. The implementations could define their own logic about how to fast copy/duplicate files, e.g. hard link or transfer manager.
2. copyFiles is not an atomic operation, How could we handle the situation when some partial files fail ? Could we return the list of successful files then the caller could decide to retry or just know them ? On Thu, May 9, 2024 at 3:46 PM Keith Lee <leekeiabstract...@gmail.com> wrote: > Hi Piotr, > > Thank you for the proposal. Looks great. > > Along similar line of Aleks' question on memory usage. > > The proposal mentions that s5cmd utilises 100% of CPU similar to Flink > 1.18. However, this will be a native process outside of the JVM. Are there > risk of large/long state download starving the TM of CPU cycle causing > issues such as heartbeat or ask timeout? > > Do you know if there is a way to limit the CPU utilisation of s5cmd? I see > worker and concurrency configuration but these do not map directly to cap > in CPU usage. The experience for feature user in this case will be one of > trial and error. > > Thanks > Keith > > On Wed, May 8, 2024 at 12:47 PM Ahmed Hamdy <hamdy10...@gmail.com> wrote: > > > Hi Piotr > > +1 for the proposal, it seems to have a lot of gains. > > > > Best Regards > > Ahmed Hamdy > > > > > > On Mon, 6 May 2024 at 12:06, Zakelly Lan <zakelly....@gmail.com> wrote: > > > > > Hi Piotrek, > > > > > > Thanks for your answers! > > > > > > Good question. The intention and use case behind > `DuplicatingFileSystem` > > is > > > > different. It marks if `FileSystem` can quickly copy/duplicate files > > > > in the remote `FileSystem`. For example an equivalent of a hard link > or > > > > bumping a reference count in the remote system. That's a bit > different > > > > to copy paths between remote and local file systems. > > > > > > > > However, it could arguably be unified under one interface where we > > would > > > > re-use or re-name `canFastDuplicate(Path, Path)` to > > > > `canFastCopy(Path, Path)` with the following use cases: > > > > - `canFastCopy(remoteA, remoteB)` returns true - current equivalent > of > > > > `DuplicatingFileSystem` - quickly duplicate/hard link remote path > > > > - `canFastCopy(local, remote)` returns true - FS can natively upload > > > local > > > > file to a remote location > > > > - `canFastCopy(remote, local)` returns true - FS can natively > download > > > > local file from a remote location > > > > > > > > Maybe indeed that's a better solution vs having two separate > interfaces > > > for > > > > copying and duplicating? > > > > > > > > > > I'd prefer a unified one interface, `canFastCopy(Path, Path)` looks > good > > to > > > me. This also resolves my question 1 about the destination. > > > > > > > > > Best, > > > Zakelly > > > > > > On Mon, May 6, 2024 at 6:36 PM Piotr Nowojski <pnowoj...@apache.org> > > > wrote: > > > > > > > Hi All! > > > > > > > > Thanks for your comments. > > > > > > > > Muhammet and Hong, about the config options. > > > > > > > > > Could you please also add the configuration property for this? An > > > example > > > > showing how users would set this parameter would be helpful. > > > > > > > > > 1/ Configure the implementation of PathsCopyingFileSystem used > > > > > 2/ Configure the location of the s5cmd binary (version control > etc.) > > > > > > > > Ops, sorry I added the config options that I had in mind to the > FLIP. I > > > > don't know why I have omitted this. Basically I suggest that in order > > to > > > > use native file copying: > > > > 1. `FileSystem` must support it via implementing > > `PathsCopyingFileSystem` > > > > interface > > > > 2. That `FileSystem` would have to be configured to actually use it. > > For > > > > example S3 file system would return `true` that it can copy paths > > > > only if `s3.s5cmd.path` has been specified. > > > > > > > > > Would this affect any filesystem connectors that use > FileSystem[1][2] > > > > dependencies? > > > > > > > > Definitely not out of the box. Any place in Flink that is currently > > > > uploading/downloading files from a FileSystem could use this feature, > > but > > > > it > > > > would have to be implemented. The same way this FLIP will implement > > > native > > > > files copying when downloading state during recovery, > > > > but the old code path will be still used for uploading state files > > > during a > > > > checkpoint. > > > > > > > > > How adding a s5cmd will affect memory footprint? Since this is a > > native > > > > binary, memory consumption will not be controlled by JVM or Flink. > > > > > > > > As you mentioned the memory usage of `s5cmd` will not be controlled, > so > > > the > > > > memory footprint will grow. S5cmd integration with Flink > > > > has been tested quite extensively on our production environment > > already, > > > > and we haven't observed any issues so far despite the fact we > > > > are using quite small pods. But of course if your setup is working on > > the > > > > edge of OOM, this could tip you over that edge. > > > > > > > > Zakelly: > > > > > > > > > 1. What is the semantic of `canCopyPath`? Should it be associated > > with > > > a > > > > > specific destination path? e.g. It can be copied to local, but not > to > > > the > > > > > remote FS. > > > > > > > > For the S3 (both for SDKv2 and s5cmd implementations), the copying > > > > direction (upload/download) doesn't matter. I don't know about other > > > > file systems, I haven't investigated anything besides S3. > Nevertheless > > I > > > > wouldn't worry too much about it, since we can start with the simple > > > > `canCopyPath` that handles both directions. If this will become > > important > > > > in the future, adding directional `canDownloadPath` or > `canUploadPath` > > > > would be a backward compatible change, so we can safely extend it in > > the > > > > future if needed. > > > > > > > > > 2. Is the existing interface `DuplicatingFileSystem` > feasible/enough > > > for > > > > this case? > > > > > > > > Good question. The intention and use case behind > > `DuplicatingFileSystem` > > > is > > > > different. It marks if `FileSystem` can quickly copy/duplicate files > > > > in the remote `FileSystem`. For example an equivalent of a hard link > or > > > > bumping a reference count in the remote system. That's a bit > different > > > > to copy paths between remote and local file systems. > > > > > > > > However, it could arguably be unified under one interface where we > > would > > > > re-use or re-name `canFastDuplicate(Path, Path)` to > > > > `canFastCopy(Path, Path)` with the following use cases: > > > > - `canFastCopy(remoteA, remoteB)` returns true - current equivalent > of > > > > `DuplicatingFileSystem` - quickly duplicate/hard link remote path > > > > - `canFastCopy(local, remote)` returns true - FS can natively upload > > > local > > > > file to a remote location > > > > - `canFastCopy(remote, local)` returns true - FS can natively > download > > > > local file from a remote location > > > > > > > > Maybe indeed that's a better solution vs having two separate > interfaces > > > for > > > > copying and duplicating? > > > > > > > > > 3. Will the interface extracting introduce a break change? > > > > > > > > No. The signature of the existing abstract `FileSystem` class would > > > remain > > > > the same. Only most/all of the abstract methods would be > > > > pulled out to the interface and abstract `FileSystem` would implement > > > that > > > > new interface. > > > > > > > > Best, > > > > Piotrek > > > > > > > > pon., 6 maj 2024 o 04:55 Zakelly Lan <zakelly....@gmail.com> > > napisał(a): > > > > > > > > > Hi Piotr, > > > > > > > > > > Thanks for the proposal. It's meaningful to speed up the state > > > download. > > > > I > > > > > get into some questions: > > > > > > > > > > 1. What is the semantic of `canCopyPath`? Should it be associated > > with > > > a > > > > > specific destination path? e.g. It can be copied to local, but not > to > > > the > > > > > remote FS. > > > > > 2. Is the existing interface `DuplicatingFileSystem` > feasible/enough > > > for > > > > > this case? > > > > > 3. Will the interface extracting introduce a break change? > > > > > > > > > > > > > > > Best, > > > > > Zakelly > > > > > > > > > > > > > > > On Thu, May 2, 2024 at 6:50 PM Aleksandr Pilipenko < > z3d...@gmail.com > > > > > > > > wrote: > > > > > > > > > > > Hi Piotr, > > > > > > > > > > > > Thanks for the proposal. > > > > > > How adding a s5cmd will affect memory footprint? Since this is a > > > native > > > > > > binary, memory consumption will not be controlled by JVM or > Flink. > > > > > > > > > > > > Thanks, > > > > > > Aleksandr > > > > > > > > > > > > On Thu, 2 May 2024 at 11:12, Hong Liang <h...@apache.org> wrote: > > > > > > > > > > > > > Hi Piotr, > > > > > > > > > > > > > > Thanks for the FLIP! Nice to see work to improve the filesystem > > > > > > > performance. +1 to future work to improve the upload speed as > > well. > > > > > This > > > > > > > would be useful for jobs with large state and high Async > > > > checkpointing > > > > > > > times. > > > > > > > > > > > > > > Some thoughts on the configuration, it might be good for us to > > > > > introduce > > > > > > 2x > > > > > > > points of configurability for future proofing: > > > > > > > 1/ Configure the implementation of PathsCopyingFileSystem used, > > > maybe > > > > > by > > > > > > > config, or by ServiceResources (this would allow us to use this > > for > > > > > > > alternative clouds/Implement S3 SDKv2 support if we want this > in > > > the > > > > > > > future). Also this could be used as a feature flag to determine > > if > > > we > > > > > > > should be using this new native file copy support. > > > > > > > 2/ Configure the location of the s5cmd binary (version control > > > etc.), > > > > > as > > > > > > > you have mentioned in the FLIP. > > > > > > > > > > > > > > Regards, > > > > > > > Hong > > > > > > > > > > > > > > > > > > > > > On Thu, May 2, 2024 at 9:40 AM Muhammet Orazov > > > > > > > <mor+fl...@morazow.com.invalid> wrote: > > > > > > > > > > > > > > > Hey Piotr, > > > > > > > > > > > > > > > > Thanks for the proposal! It would be great improvement! > > > > > > > > > > > > > > > > Some questions from my side: > > > > > > > > > > > > > > > > > In order to configure s5cmd Flink’s user would need > > > > > > > > > to specify path to the s5cmd binary. > > > > > > > > > > > > > > > > Could you please also add the configuration property > > > > > > > > for this? An example showing how users would set this > > > > > > > > parameter would be helpful. > > > > > > > > > > > > > > > > Would this affect any filesystem connectors that use > > > > > > > > FileSystem[1][2] dependencies? > > > > > > > > > > > > > > > > Best, > > > > > > > > Muhammet > > > > > > > > > > > > > > > > [1]: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/filesystems/s3/ > > > > > > > > [2]: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/filesystem/ > > > > > > > > > > > > > > > > On 2024-04-30 13:15, Piotr Nowojski wrote: > > > > > > > > > Hi all! > > > > > > > > > > > > > > > > > > I would like to put under discussion: > > > > > > > > > > > > > > > > > > FLIP-444: Native file copy support > > > > > > > > > https://cwiki.apache.org/confluence/x/rAn9EQ > > > > > > > > > > > > > > > > > > This proposal aims to speed up Flink recovery times, by > > > speeding > > > > up > > > > > > > > > state > > > > > > > > > download times. However in the future, the same mechanism > > could > > > > be > > > > > > also > > > > > > > > > used to speed up state uploading > > (checkpointing/savepointing). > > > > > > > > > > > > > > > > > > I'm curious to hear your thoughts. > > > > > > > > > > > > > > > > > > Best, > > > > > > > > > Piotrek > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > -- Best, Hangxiang.