[FLINK-35739] FLIP-444: Native file copy support

• Type: New Feature
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 2.0-preview
• Component/s: Connectors / FileSystem, Runtime / State Backends
• Labels: pull-request-available

Release Note

Users can now configure Flink to use s5cmd to speed up downloading files from S3 during the recovery process, when using RocksDB, by a factor of 2.

(link the docs)

• Assignee: Piotr Nowojski
• Reporter: Piotr Nowojski
• Created: 02/Jul/24 08:37
• Updated: 02/Sep/24 09:35
• Resolved: 28/Aug/24 08:23

Description

https://cwiki.apache.org/confluence/display/FLINK/FLIP-444%3A+Native+file+copy+support

State downloading in Flink can be a time and CPU consuming operation, which is especially visible if CPU resources per task slot are strictly restricted to for example a single CPU. Downloading 1GB of state size can take significant amount of time, while the code doing so is quite inefficient.

Currently when downloading state files, Flink is creating an FSDataInputStream from the remote file, and copies its bytes, to an OutputStream pointing to a local file (in the RocksDBStateDownloader#downloadDataForStateHandle method). FSDataInputStream internally is being wrapped by many layers of abstractions and indirections and what’s worse, every file is being copied individually, which leads to quite high overheads for small files. Download times and download process CPU efficiency can be significantly improved if we introduced an API to allow org.apache.flink.core.fs.FileSystem to copy many files natively and all at once.

For S3, there are at least two potential implementations. The first one is using AWS SDKv2 directly (Flink currently is using AWS SDKv1 wrapped by hadoop/presto) and Amazon S3 Transfer Manager. Second option is to use a 3rd party tool called s5cmd. It is claimed to be a faster alternative to the official AWS clients, which was confirmed by our benchmarks.

Issue Links

• links to: GitHub Pull Request #25028

Sub-Tasks

• Provide native file copy support for S3 using s5cmd: Closed
• Use native file copy in RocksDBStateDownloader: Closed
• Interrupt s5cmd call on cancellation: Closed
• Limit s5cmd resource usage: Closed
• Deprecate/remove DuplicatingFileSystem: Closed
• Document s5cmd: Closed
• Wait for state download on cancellation to enforce cleanup: Closed