[FLINK-35768][FLIP-444] Native file copy support #25028

pnowojski · 2024-07-05T16:27:56Z

What is the purpose of the change

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.

Brief change log

This PR covers two first steps for FLIP-444:

Verifying this change

This change is covered by a couple of new ITCases, that run s5cmd against Minio and some unit tests.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2024-07-05T16:31:35Z

CI report:

4671268 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

rkhachatryan

Thanks for the PR, LGTM in general.
I've left some comments, PTAL.

(disclaimer: I've reviewed the internal version of this PR already)

rkhachatryan · 2024-07-08T13:23:14Z

...link-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/AbstractS3FileSystemFactory.java

+    public static final ConfigOption<String> ACCESS_KEY =
+            ConfigOptions.key("s3.access-key")


I couldn't find generated docs for these options (all s3 options).
Should we start generating in this hotfix commit?

Should we start generating in this hotfix commit?

Probably a good idea, but I guess that would require some larger re-work of the docs. I presume currently the docs, including the list of configuration parameters, are manually crafted, without any part that is automatically generated.

My motivation was to have those ConfigOptions in one place here, to avoid copy/pasting the keys in the code various of places, including tests.

flink-core/src/main/java/org/apache/flink/util/FileUtils.java

...esystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/FlinkS3FileSystem.java

rkhachatryan · 2024-07-08T13:31:43Z

...systems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/MinioTestContainer.java

    public void setS3ConfigOptions(Configuration config) {
-        config.setString(FLINK_CONFIG_S3_ENDPOINT, getHttpEndpoint());
+        config.set(AbstractS3FileSystemFactory.ENDPOINT, getHttpEndpoint());


nit: re-order commits so that the use of options goes after their introduction?

Currently, I see it as

[hotfix] Use newly defined ConfigOptions in MinioTestContainer [hotfix] Move CompressionUtils to flink-core [hotfix] Create ConfigOptions for s3 access/secret keys and endpoint

Huh? For me this ("[hotfix] Use newly defined ConfigOptions in MinioTestContainer") is the 3rd commit

rkhachatryan · 2024-07-08T13:32:31Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/StreamStateHandle.java

+    default Optional<org.apache.flink.core.fs.Path> maybeGetPath() {
+        return Optional.empty();
+    }


Should this also be implemented by DirectoryStreamStateHandle and SegmentFileStateHandle?

It looks like DirectoryStreamStateHandle can not be used for recovery any way due to:

public FSDataInputStream openInputStream() { throw new UnsupportedOperationException(); }

Also in that case I think we would have to add a special handling of directories (The s5cmd docs do not state that clearly, but it looks like for download, and only for download, you need to add a wildcard *?)

Direct copy files for SegmentFileStateHandle might be less efficient, if roughly speaking, the desired segment is < 50% of the underlying file. So I'm not sure if I would use it. Maybe in the future we should add some code letting the filesystem to decide if it's worth copying given state handle or not? 🤔 Moving more logic of deciding how to download handles from the RocksDBDownloader to FileSystem?

I was thinking about other potential usages of maybeGetPath (not s5cmd), e.g. in tests. I'd assume that these handles would return some path, and returning empty would break this assumption.
And this wouldn't necessarily mean that s5 will be used for these handles - it also depends on the caller (rocksdb downloader).
Anyways, this would be a NIT, so please feel free to leave it as is.

Ok, I see.

And this wouldn't necessarily mean that s5 will be used for these handles - it also depends on the caller (rocksdb downloader).

I somehow have a feeling that in the future we might want to rethink how to decide which downloading/uploading code path to use.

...d-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateDownloader.java

pnowojski

I've updated the PR, can you take a look again @rkhachatryan ?

pnowojski · 2024-07-08T14:54:33Z

...link-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/AbstractS3FileSystemFactory.java

+    public static final ConfigOption<String> ACCESS_KEY =
+            ConfigOptions.key("s3.access-key")


Should we start generating in this hotfix commit?

Probably a good idea, but I guess that would require some larger re-work of the docs. I presume currently the docs, including the list of configuration parameters, are manually crafted, without any part that is automatically generated.

My motivation was to have those ConfigOptions in one place here, to avoid copy/pasting the keys in the code various of places, including tests.

flink-core/src/main/java/org/apache/flink/util/FileUtils.java

...esystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/FlinkS3FileSystem.java

pnowojski · 2024-07-09T08:34:00Z

...systems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/MinioTestContainer.java

    public void setS3ConfigOptions(Configuration config) {
-        config.setString(FLINK_CONFIG_S3_ENDPOINT, getHttpEndpoint());
+        config.set(AbstractS3FileSystemFactory.ENDPOINT, getHttpEndpoint());


Huh? For me this ("[hotfix] Use newly defined ConfigOptions in MinioTestContainer") is the 3rd commit

pnowojski · 2024-07-09T09:07:18Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/StreamStateHandle.java

+    default Optional<org.apache.flink.core.fs.Path> maybeGetPath() {
+        return Optional.empty();
+    }


It looks like DirectoryStreamStateHandle can not be used for recovery any way due to:

public FSDataInputStream openInputStream() { throw new UnsupportedOperationException(); }

Also in that case I think we would have to add a special handling of directories (The s5cmd docs do not state that clearly, but it looks like for download, and only for download, you need to add a wildcard *?)

Direct copy files for SegmentFileStateHandle might be less efficient, if roughly speaking, the desired segment is < 50% of the underlying file. So I'm not sure if I would use it. Maybe in the future we should add some code letting the filesystem to decide if it's worth copying given state handle or not? 🤔 Moving more logic of deciding how to download handles from the RocksDBDownloader to FileSystem?

rkhachatryan

LGTM

…ownloader

pnowojski requested a review from rkhachatryan July 5, 2024 16:28

pnowojski force-pushed the f35768 branch 3 times, most recently from b58fdcb to 62536a3 Compare July 5, 2024 16:48

flinkbot added component=Connectors/FileSystem component=Runtime/StateBackends labels Jul 5, 2024

pnowojski force-pushed the f35768 branch from 62536a3 to b1cc48e Compare July 8, 2024 07:40

rkhachatryan reviewed Jul 8, 2024

View reviewed changes

pnowojski commented Jul 9, 2024

View reviewed changes

rkhachatryan approved these changes Jul 10, 2024

View reviewed changes

pnowojski changed the title ~~[FLINK-35739][FLIP-444] Native file copy support~~ [FLINK-35768][FLIP-444] Native file copy support Jul 10, 2024

pnowojski force-pushed the f35768 branch from b1cc48e to 85aa2e6 Compare July 10, 2024 16:08

flinkbot added component=<none> and removed component=Connectors/FileSystem component=Runtime/StateBackends labels Jul 10, 2024

pnowojski force-pushed the f35768 branch from 85aa2e6 to 804cf3c Compare July 11, 2024 12:11

pnowojski added 8 commits July 11, 2024 17:08

[hotfix] Create ConfigOptions for s3 access/secret keys and endpoint

33b0de9

[hotfix] Move CompressionUtils to flink-core

3bac7fc

[hotfix] Use newly defined ConfigOptions in MinioTestContainer

5648cf6

[FLINK-35767][filesystems] Add PathsCopyingFileSystem

ba33b46

[FLINK-35767][s3] Use s5cmd for fast files download from the s3

adf0799

[FLINK-35768][state,rocksdb] Refactor RocksDBStateDownloader

fac4be7

[FLINK-35768][state,rocksdb] Use fast files download in RocksdBStateD…

3731b43

…ownloader

[FLINK-35768][tests] Add an ITCase for s5cmd

4671268

pnowojski force-pushed the f35768 branch from 804cf3c to 4671268 Compare July 11, 2024 15:08

pnowojski merged commit a056e02 into apache:master Jul 12, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-35768][FLIP-444] Native file copy support #25028

[FLINK-35768][FLIP-444] Native file copy support #25028

pnowojski commented Jul 5, 2024

flinkbot commented Jul 5, 2024 •

edited

Loading

rkhachatryan left a comment

rkhachatryan Jul 8, 2024

pnowojski Jul 8, 2024

rkhachatryan Jul 8, 2024

pnowojski Jul 9, 2024

rkhachatryan Jul 8, 2024

pnowojski Jul 9, 2024

rkhachatryan Jul 10, 2024

pnowojski Jul 10, 2024

pnowojski left a comment

pnowojski Jul 8, 2024

pnowojski Jul 9, 2024

pnowojski Jul 9, 2024

rkhachatryan left a comment

		public static final ConfigOption<String> ACCESS_KEY =
		ConfigOptions.key("s3.access-key")

[FLINK-35768][FLIP-444] Native file copy support #25028

[FLINK-35768][FLIP-444] Native file copy support #25028

Conversation

pnowojski commented Jul 5, 2024

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Jul 5, 2024 • edited Loading

CI report:

rkhachatryan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pnowojski left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rkhachatryan left a comment

Choose a reason for hiding this comment

flinkbot commented Jul 5, 2024 •

edited

Loading