You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/11/03 13:53:49 UTC
[GitHub] [flink] 1996fanrui opened a new pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
1996fanrui opened a new pull request #13885:
URL: https://github.com/apache/flink/pull/13885
## What is the purpose of the change
Read checkpoint stream with buffer to speedup restore.
There will be many small IOs in the restore process, frequent small io is not friendly to disk. So add buffer to improve IO performance.
## Brief change log
- add config `state.backend.fs.read-buffer-size`
- Design FSDataBufferedInputStream, FSDataInputStream with buffer
- Use FSDataBufferedInputStream to wrap fsDataInputStream in HeapRestoreOperation#restore.
tips:
The design of FSDataBufferedInputStream refers to FsCheckpointStateOutputStream and Java's BufferedInputStream.
Compared to Java's BufferedInputStream, the seek and getPos interfaces are mainly added.
## Verifying this change
This change added tests and can be verified as follows:
- Added order read for FSDataBufferedInputStream
- Added seek and random read for FSDataBufferedInputStream
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: yes
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? JavaDocs
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-735033850
> Hi @1996fanrui ,
>
> that's an interesting discovery and investigation that you did there!
>
> I think the approach on the filesystem level is also much better than the previous way. Let's try to not change any public API (FileSystem) as this would slow down the progress.
>
> I'd probably focus on Hadoop file systems entirely (for now). What I'd propose is the following:
>
> * Use `HadoopFsFactory#configure` to extract the buffer size and pass it to the `ctor` of all filesystems created by the factory.
> * Use that default buffer size in `HadoopFileSystem#open(Path)` to call `#open(Path, int)`.
> * `HadoopFileSystem#open(Path, int)` should use the buffer size both in the call to Hadoop and to wrap it as you did as in the `BufferedFSInputStream`. I dug a bit into the Hadoop code and noticed that the cache is by default just 4kb. So even if we have cache on top of it with 64kb, we would still need to ask Hadoop several times.
>
> So, that means you are not adding any new methods, but just modify existing ones.
Hi @Myasuka @AHeise ,
Based on the above design, there are some questions that I hope to be answered:
- `buffer size` is passed to HadoopFileSystem through Constructor. It means that HadoopFileSystem needs to add a new Constructor: `HadoopFileSystem(FileSystem, bufferSize)`. HadoopFsFactory will call the new constructor.
- The old constructor `HadoopFileSystem(FileSystem)` is still called by other FsFactory. So other FsFactory will not be able to improve performance. For example: OSSFileSystemFactory.
Question: Do other FsFactory related to HadoopFileSystem need to improve performance? If needed, are there other better designs. (Can there be a way not to modify many FsFactory?)
Thanks.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #13885:
URL: https://github.com/apache/flink/pull/13885#discussion_r517888118
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FSDataBufferedInputStream.java
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.state.filesystem;
+
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * FSDataInputStream with Buffer
+ */
+public class FSDataBufferedInputStream extends FSDataInputStream {
Review comment:
I think most of your code is copied from `java.io.BufferedInputStream`, right? Please add notes of this in the javadoc.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FSDataBufferedInputStream.java
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.state.filesystem;
+
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * FSDataInputStream with Buffer
+ */
+public class FSDataBufferedInputStream extends FSDataInputStream {
+
+ private static final int DEFAULT_BUFFER_SIZE = 8192;
+
+ protected volatile byte[] buf;
+
+ // read offset of buf
+ private int pos;
+
+ // availed count of buf
+ private int count;
+
+ private final FSDataInputStream inputStream;
+
+ private volatile boolean closed;
+
+ public FSDataBufferedInputStream(FSDataInputStream inputStream) {
+ this(inputStream, DEFAULT_BUFFER_SIZE);
+ }
+
+ public FSDataBufferedInputStream(
+ FSDataInputStream inputStream,
+ int bufferSize) {
+ this.inputStream = inputStream;
+
+ Preconditions.checkState(bufferSize > 0,
+ new IllegalArgumentException("bufferSize must > 0"));
Review comment:
We should pass error message instead of `IllegalArgumentException` here.
##########
File path: flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
##########
@@ -157,4 +157,14 @@
.withDescription(String.format("The default size of the write buffer for the checkpoint streams that write to file systems. " +
"The actual write buffer size is determined to be the maximum of the value of this option and option '%s'.", FS_SMALL_FILE_THRESHOLD.key()));
+ /**
+ * The default size of the read buffer for the checkpoint streams that read from file systems.
+ */
+ @Documentation.Section(Documentation.Sections.EXPERT_STATE_BACKENDS)
+ public static final ConfigOption<Integer> FS_READ_BUFFER_SIZE = ConfigOptions
+ .key("state.backend.fs.read-buffer-size")
+ .intType()
Review comment:
I am also in favor of `memory type` as the buffer is just an array of bytes allocated from JVM memory. The reason why `state.backend.fs.write-buffer-size` is integer type is that it implemented before Flink introduced `memory type` configuration.
##########
File path: flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
##########
@@ -877,6 +913,20 @@ public long getWriteBatchSize() {
WRITE_BATCH_SIZE.defaultValue().getBytes() : writeBatchSize;
}
+ /**
+ * Gets the read buffer size for read checkpoint stream.
+ *
+ * <p>If not explicitly configured, this is the default value of
+ * {@link CheckpointingOptions#FS_READ_BUFFER_SIZE}.
+ *
+ * @return The file system read buffer size, in bytes.
+ */
+ public int getFsReadBufferSize() {
Review comment:
As you can see, there existed three duplicate `getFsReadBufferSize` in your PR, which could be an abstract base method for all state backends.
If we think it one step further, we now has same interfaces for checkpoint write of state backend, e.g. `createCheckpointStorage`, however, we do not have same interfaces for checkpoint **read** of state backend. That's why you meet such embarrassment here.
I think this deserve a discussion over FLINK-19465 and FLIP-142, @sjwiesman what do you think of this?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FSDataBufferedInputStream.java
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.state.filesystem;
+
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * FSDataInputStream with Buffer
+ */
+public class FSDataBufferedInputStream extends FSDataInputStream {
+
+ private static final int DEFAULT_BUFFER_SIZE = 8192;
+
+ protected volatile byte[] buf;
Review comment:
Unlike `java.io.BufferedInputStream`, `FSDataBufferedInputStream` should only be used by single thread, there is no need to make `buf` as `volatile`.
##########
File path: flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
##########
@@ -264,12 +270,36 @@ public RocksDBStateBackend(StateBackend checkpointStreamBackend) {
* @param enableIncrementalCheckpointing True if incremental checkpointing is enabled.
*/
public RocksDBStateBackend(StateBackend checkpointStreamBackend, TernaryBoolean enableIncrementalCheckpointing) {
+ this(checkpointStreamBackend, enableIncrementalCheckpointing, -1);
+ }
+
+ /**
+ * Creates a new {@code RocksDBStateBackend} that uses the given state backend to store its
+ * checkpoint data streams. Typically, one would supply a filesystem or database state backend
+ * here where the snapshots from RocksDB would be stored.
+ *
+ * <p>The snapshots of the RocksDB state will be stored using the given backend's
+ * {@link StateBackend#createCheckpointStorage(JobID)}.
+ *
+ * @param checkpointStreamBackend The backend write the checkpoint streams to.
+ * @param enableIncrementalCheckpointing True if incremental checkpointing is enabled.
+ * @param fsReadBufferSize The read buffer size for read checkpoint stream.
+ */
+ public RocksDBStateBackend(
+ StateBackend checkpointStreamBackend,
+ TernaryBoolean enableIncrementalCheckpointing,
+ int fsReadBufferSize) {
Review comment:
I don't think we should add a new constructor here. Putting `fsReadBufferSize` as one of the property in `FsStateBackend` is enough.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-731590386
> Yes, I think that makes sense. Adding a `fs.buffer` config option, maybe with a default value of 64k?
>
> * For local streams, it would be good as well, that is true.
> * For S3, we get that implicitly if we buffer the `HadoopDataInputStream`, because the S3 reads go through some Hadoop utilities and also wrap the stream with the `HadoopDataInputStream` in the end.
@StephanEwen Thanks for your reply.
After the above analysis, I think `FSDataBufferedInputStream` can wrap all InputStream.
Now there is a question: we do not have same interfaces for checkpoint read of state backend. How to pass `buffer size`?
There are three options:
1. Wait for the completion of [FLINK-19465](https://issues.apache.org/jira/browse/FLINK-19465), and then continue the current PR work
2. The current PR does not support modification of `buffer size`, and is set to 64KB uniformly.
3. Use three duplicate `getFsReadBufferSize` to complete the current PR, and then refactor in [FLINK-19465](https://issues.apache.org/jira/browse/FLINK-19465), or create a new issue to refactor.
what do you think of this? @Myasuka @sjwiesman
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-725094794
@StephanEwen Do you have any ideas? Hope to get your guidance.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] klion26 commented on a change in pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
klion26 commented on a change in pull request #13885:
URL: https://github.com/apache/flink/pull/13885#discussion_r516440861
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FSDataBufferedInputStreamTest.java
##########
@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.state.filesystem;
+
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.runtime.state.StreamStateHandle;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.UUID;
+
+/**
+ * Unit tests for the {@link FSDataBufferedInputStream}.
+ */
+public class FSDataBufferedInputStreamTest {
+
+ private static final int KB = 1024;
+ private static final int MB = 1024 * KB;
+
+ FsCheckpointStreamFactory.FsCheckpointStateOutputStream outputStream;
+ DataOutputViewStreamWrapper outView;
+
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Before
+ public void before() throws IOException {
+ String path = temporaryFolder.newFolder().getAbsolutePath();
+ File outFile = new File(path, String.valueOf(UUID.randomUUID()));
+
+ Path outPath = new Path(outFile.toURI());
+ outputStream = new FsCheckpointStreamFactory.FsCheckpointStateOutputStream(
+ outPath, outPath.getFileSystem(), MB, KB);
+ outView = new DataOutputViewStreamWrapper(outputStream);
+ }
+
+ @Test
+ public void testOrderRead() throws Exception {
+ // write
+ int num = 10_000_000;
+ StreamStateHandle stateHandle = writeData(num);
+ Assert.assertNotNull(stateHandle);
+
+ int expectedStateSize = 2 * num * Integer.BYTES;
+ Assert.assertEquals(expectedStateSize, stateHandle.getStateSize());
+
+ // start read
+ FSDataInputStream fsDataInputStream = stateHandle.openInputStream();
+ FSDataBufferedInputStream bufferedInputStream = new FSDataBufferedInputStream(fsDataInputStream);
+ DataInputViewStreamWrapper inputView = new DataInputViewStreamWrapper(bufferedInputStream);
+
+ for (int i = 1; i <= num; i++) {
+ int read = inputView.readInt();
+ Assert.assertEquals(i, read);
+ }
+ for (int i = num; i >= 1; i--) {
+ int read = inputView.readInt();
+ Assert.assertEquals(i, read);
+ }
+ bufferedInputStream.close();
+ stateHandle.discardState();
+ }
+
+ @Test
+ public void testSeek() throws Exception {
Review comment:
Do we need to add some bound test cases for `seek()` (such as seek out of bounding) here?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui closed pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
1996fanrui closed pull request #13885:
URL: https://github.com/apache/flink/pull/13885
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-724092762
> Are the HDFS input streams generally not buffered? Would it make sense to adjust the `HadoopDataInputStream` class to be buffered?
Hi @StephanEwen , I am sorry to reply you so late.
In the past two days, I read the relevant code of hdfs reading data and did some performance analysis.
conclusion as below:
The hdfs input stream has buffer by default. The default buffer size is 64KB.
In theory, Use FSDataBufferedInputStream to wrap hdfsInputStream does not reduce the number of disk accesses. But the result shows: When buffer is added, restore time is reduced to one-third of the original.
So, I did some performance analysis.
From the point of view of CPU usage, When buffer is added cpu usage is greatly reduced.
Using hdfsInputStream directly, the CPU usage rate is 60~70%.
Use FSDataBufferedInputStream to wrap hdfsInputStream, the CPU usage rate is 20~25%.
Analyze why hdfsInputStream consumes CPU:
The hdfs client contains a lot of statistical information, and the method call stack is relatively deep, each method will consume a little performance. As shown in the picture below:
[flame graph remark link](https://drive.google.com/file/d/1zTwHdmSybAgyBGIIP71FLfMQ5DvK1eON/view?usp=sharing)
If you use FSDataBufferedInputStream to wrap hdfsInputStream, there will be a buffer outside the hdfs client, avoiding a very deep call stack and a lot of hdfs statistics.
Original flame graph link:
[restore with buffer](https://drive.google.com/file/d/1jxBNzh2iIsrX__wfFjvTWiC9LPCCDS3A/view?usp=sharing)
[restore without buffer](https://drive.google.com/file/d/1qDbqQC4bG34_ZsCrOpDaNouNQpvVXRWH/view?usp=sharing)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on a change in pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on a change in pull request #13885:
URL: https://github.com/apache/flink/pull/13885#discussion_r518475872
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FSDataBufferedInputStream.java
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.state.filesystem;
+
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * FSDataInputStream with Buffer
+ */
+public class FSDataBufferedInputStream extends FSDataInputStream {
+
+ private static final int DEFAULT_BUFFER_SIZE = 8192;
+
+ protected volatile byte[] buf;
+
+ // read offset of buf
+ private int pos;
+
+ // availed count of buf
+ private int count;
+
+ private final FSDataInputStream inputStream;
+
+ private volatile boolean closed;
+
+ public FSDataBufferedInputStream(FSDataInputStream inputStream) {
+ this(inputStream, DEFAULT_BUFFER_SIZE);
+ }
+
+ public FSDataBufferedInputStream(
+ FSDataInputStream inputStream,
+ int bufferSize) {
+ this.inputStream = inputStream;
+
+ Preconditions.checkState(bufferSize > 0,
+ new IllegalArgumentException("bufferSize must > 0"));
Review comment:
done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on a change in pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on a change in pull request #13885:
URL: https://github.com/apache/flink/pull/13885#discussion_r517964413
##########
File path: flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
##########
@@ -877,6 +913,20 @@ public long getWriteBatchSize() {
WRITE_BATCH_SIZE.defaultValue().getBytes() : writeBatchSize;
}
+ /**
+ * Gets the read buffer size for read checkpoint stream.
+ *
+ * <p>If not explicitly configured, this is the default value of
+ * {@link CheckpointingOptions#FS_READ_BUFFER_SIZE}.
+ *
+ * @return The file system read buffer size, in bytes.
+ */
+ public int getFsReadBufferSize() {
Review comment:
Thanks to @Myasuka 's perfect analysis, this is indeed the reason for my design.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-720452604
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a53411ad7bc59231419ea53bbcbfe3fb95ff67b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8794",
"triggerID" : "a53411ad7bc59231419ea53bbcbfe3fb95ff67b7",
"triggerType" : "PUSH"
}, {
"hash" : "91c08c8b38d94746603022eb07f85c3b4e3c095e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8835",
"triggerID" : "91c08c8b38d94746603022eb07f85c3b4e3c095e",
"triggerType" : "PUSH"
}, {
"hash" : "30162c89af86d6e2e666f9e119e98e7cf91548d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8839",
"triggerID" : "30162c89af86d6e2e666f9e119e98e7cf91548d3",
"triggerType" : "PUSH"
}, {
"hash" : "461553441634d634eb8950a60999a0eebc21b3dd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8853",
"triggerID" : "461553441634d634eb8950a60999a0eebc21b3dd",
"triggerType" : "PUSH"
}, {
"hash" : "8872b60767d3e3ddf942564e1df3d2bb325fff6b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9086",
"triggerID" : "8872b60767d3e3ddf942564e1df3d2bb325fff6b",
"triggerType" : "PUSH"
}, {
"hash" : "be7f3c7a5db0320f94ae49c57cb72f56c477bcdd",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9902",
"triggerID" : "be7f3c7a5db0320f94ae49c57cb72f56c477bcdd",
"triggerType" : "PUSH"
}, {
"hash" : "4f972fcfd74e2d74c5cc7af7cc8561faadd21d1f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4f972fcfd74e2d74c5cc7af7cc8561faadd21d1f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* be7f3c7a5db0320f94ae49c57cb72f56c477bcdd Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9902)
* 4f972fcfd74e2d74c5cc7af7cc8561faadd21d1f UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-734116774
> > > `LocalFileSystem`
> >
> >
> > Yes, it looks like `LocalFileSystem` should also be a good candidate. In particular `#open(Path, int)` currently ignores the buffer size...
> > I'm wondering if it makes more sense to create two PRs though. Even though the solution is pretty much the same, the underlying issues are different imho. The smaller a PR, the faster we can usually go. However, since they are so related, I wouldn't mind both issues being resolved in the same PR.
>
> Hi @AHeise , thanks for your quick reply.
>
> I think it is okay to create two PRs. The current PR developed `FsDataBuefferedInputStream` and applied it to `HadoopFileSysytem`.
>
> The next PR considers whether other FileSystem are profitable.
Hi @AHeise ,
Do you think this plan is OK? If ok, I will develop the code of the current PR when I have time.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
AHeise commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-732215815
Hi @1996fanrui ,
that's an interesting discovery and investigation that you did there!
I think the approach on the filesystem level is also much better than the previous way. Let's try to not change any public API (FileSystem) as this would slow down the progress.
I'd probably focus on Hadoop file systems entirely (for now). What I'd propose is the following:
- Use `HadoopFsFactory#configure` to extract the buffer size and pass it to the `ctor` of all filesystems created by the factory.
- Use that default buffer size in `HadoopFileSystem#open(Path)` to call `#open(Path, int)`.
- `HadoopFileSystem#open(Path, int)` should use the buffer size both in the call to Hadoop and to wrap it as you did as in the `BufferedFSInputStream`. I dug a bit into the Hadoop code and noticed that the cache is by default just 4kb. So even if we have cache on top of it with 64kb, we would still need to ask Hadoop several times.
So, that means you are not adding any new methods, but just modify existing ones.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-722172811
> I've left a few comments. Please remove the merge commits from this branch and check if the test failure is relevant to this change.
@sjwiesman Thanks for your review.
I have removed these merge commits and modified the initialization of config.
Please help me to review again when you are free. Thanks.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-734190473
Hi @AHeise @Myasuka ,Thanks for your reply. I will do it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-723504501
Are the HDFS input streams generally not buffered? Would it make sense to adjust the `HadoopDataInputStream` class to be buffered?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-732336693
> > `LocalFileSystem`
>
> Yes, it looks like `LocalFileSystem` should also be a good candidate. In particular `#open(Path, int)` currently ignores the buffer size...
>
> I'm wondering if it makes more sense to create two PRs though. Even though the solution is pretty much the same, the underlying issues are different imho. The smaller a PR, the faster we can usually go. However, since they are so related, I wouldn't mind both issues being resolved in the same PR.
Hi @AHeise , thanks for your quick reply.
I think it is okay to create two PRs. The current PR developed `FsDataBuefferedInputStream` and applied it to `HadoopFileSysytem`.
The next PR considers whether other FileSystem are profitable.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-720452604
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
AHeise commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-734129526
> > > > `LocalFileSystem`
> > >
> > >
> > > Yes, it looks like `LocalFileSystem` should also be a good candidate. In particular `#open(Path, int)` currently ignores the buffer size...
> > > I'm wondering if it makes more sense to create two PRs though. Even though the solution is pretty much the same, the underlying issues are different imho. The smaller a PR, the faster we can usually go. However, since they are so related, I wouldn't mind both issues being resolved in the same PR.
> >
> >
> > Hi @AHeise , thanks for your quick reply.
> > I think it is okay to create two PRs. The current PR developed `FsDataBuefferedInputStream` and applied it to `HadoopFileSysytem`.
> > The next PR considers whether other FileSystem are profitable.
>
> Hi @AHeise ,
>
> Do you think this plan is OK? If ok, I will develop the code of the current PR when I have time.
Yes @1996fanrui , please go ahead.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on a change in pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on a change in pull request #13885:
URL: https://github.com/apache/flink/pull/13885#discussion_r517959615
##########
File path: flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
##########
@@ -157,4 +157,14 @@
.withDescription(String.format("The default size of the write buffer for the checkpoint streams that write to file systems. " +
"The actual write buffer size is determined to be the maximum of the value of this option and option '%s'.", FS_SMALL_FILE_THRESHOLD.key()));
+ /**
+ * The default size of the read buffer for the checkpoint streams that read from file systems.
+ */
+ @Documentation.Section(Documentation.Sections.EXPERT_STATE_BACKENDS)
+ public static final ConfigOption<Integer> FS_READ_BUFFER_SIZE = ConfigOptions
+ .key("state.backend.fs.read-buffer-size")
+ .intType()
Review comment:
done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-733649848
@StephanEwen Thank you for your very useful suggestions in the last few days, and I wish you a happy work.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] sjwiesman commented on a change in pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
sjwiesman commented on a change in pull request #13885:
URL: https://github.com/apache/flink/pull/13885#discussion_r516245707
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
##########
@@ -524,13 +524,17 @@ public CheckpointStorageAccess createCheckpointStorage(JobID jobId) throws IOExc
HeapPriorityQueueSetFactory priorityQueueSetFactory =
new HeapPriorityQueueSetFactory(keyGroupRange, numberOfKeyGroups, 128);
+ ReadableConfig config = env.getTaskManagerInfo().getConfiguration();
+ int fsReadBufferSize = config.get(CheckpointingOptions.FS_READ_BUFFER_SIZE);
Review comment:
Do not do this here, you should use the config passed to the `configure` method. Look at how the async flag is set.
##########
File path: flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
##########
@@ -540,6 +540,10 @@ public CheckpointStorageAccess createCheckpointStorage(JobID jobId) throws IOExc
ExecutionConfig executionConfig = env.getExecutionConfig();
StreamCompressionDecorator keyGroupCompressionDecorator = getCompressionDecorator(executionConfig);
+
+ ReadableConfig config = env.getTaskManagerInfo().getConfiguration();
+ int fsReadBufferSize = config.get(CheckpointingOptions.FS_READ_BUFFER_SIZE);
Review comment:
Same
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java
##########
@@ -331,13 +331,18 @@ public OperatorStateBackend createOperatorStateBackend(
TaskStateManager taskStateManager = env.getTaskStateManager();
HeapPriorityQueueSetFactory priorityQueueSetFactory =
new HeapPriorityQueueSetFactory(keyGroupRange, numberOfKeyGroups, 128);
+
+ ReadableConfig config = env.getTaskManagerInfo().getConfiguration();
+ int fsReadBufferSize = config.get(CheckpointingOptions.FS_READ_BUFFER_SIZE);
Review comment:
Same as above, please do this inside of `configure`.
##########
File path: flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
##########
@@ -157,4 +157,14 @@
.withDescription(String.format("The default size of the write buffer for the checkpoint streams that write to file systems. " +
"The actual write buffer size is determined to be the maximum of the value of this option and option '%s'.", FS_SMALL_FILE_THRESHOLD.key()));
+ /**
+ * The default size of the read buffer for the checkpoint streams that read from file systems.
+ */
+ @Documentation.Section(Documentation.Sections.EXPERT_STATE_BACKENDS)
+ public static final ConfigOption<Integer> FS_READ_BUFFER_SIZE = ConfigOptions
+ .key("state.backend.fs.read-buffer-size")
+ .intType()
Review comment:
Shouldn't the type of this be `memory`? I know some of the others aren't but that's because they predate us having a memory configuration type.
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockStateBackend.java
##########
@@ -134,13 +136,18 @@ public CheckpointStreamFactory resolveCheckpointStorageLocation(long checkpointI
MetricGroup metricGroup,
@Nonnull Collection<KeyedStateHandle> stateHandles,
CloseableRegistry cancelStreamRegistry) {
+
+ Configuration configuration = env.getTaskManagerInfo().getConfiguration();
+ int fsReadBufferSize = configuration.getInteger(CheckpointingOptions.FS_READ_BUFFER_SIZE);
Review comment:
Same
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on a change in pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on a change in pull request #13885:
URL: https://github.com/apache/flink/pull/13885#discussion_r517932206
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FSDataBufferedInputStream.java
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.state.filesystem;
+
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * FSDataInputStream with Buffer
+ */
+public class FSDataBufferedInputStream extends FSDataInputStream {
Review comment:
Thanks for your review.
Some code is copied from `java.io.BufferedInputStream`. Compared to `java.io.BufferedInputStream`, the seek and getPos interfaces are mainly added.
I will add some notes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-732736731
Thanks @AHeise for helping out here. Sorry @1996fanrui I am pretty overloaded the last days, so I could not follow up in a timely manner.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on a change in pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on a change in pull request #13885:
URL: https://github.com/apache/flink/pull/13885#discussion_r518475796
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FSDataBufferedInputStream.java
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.state.filesystem;
+
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * FSDataInputStream with Buffer
+ */
+public class FSDataBufferedInputStream extends FSDataInputStream {
+
+ private static final int DEFAULT_BUFFER_SIZE = 8192;
+
+ protected volatile byte[] buf;
Review comment:
done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on a change in pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on a change in pull request #13885:
URL: https://github.com/apache/flink/pull/13885#discussion_r516464886
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FSDataBufferedInputStreamTest.java
##########
@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.state.filesystem;
+
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.runtime.state.StreamStateHandle;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.UUID;
+
+/**
+ * Unit tests for the {@link FSDataBufferedInputStream}.
+ */
+public class FSDataBufferedInputStreamTest {
+
+ private static final int KB = 1024;
+ private static final int MB = 1024 * KB;
+
+ FsCheckpointStreamFactory.FsCheckpointStateOutputStream outputStream;
+ DataOutputViewStreamWrapper outView;
+
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Before
+ public void before() throws IOException {
+ String path = temporaryFolder.newFolder().getAbsolutePath();
+ File outFile = new File(path, String.valueOf(UUID.randomUUID()));
+
+ Path outPath = new Path(outFile.toURI());
+ outputStream = new FsCheckpointStreamFactory.FsCheckpointStateOutputStream(
+ outPath, outPath.getFileSystem(), MB, KB);
+ outView = new DataOutputViewStreamWrapper(outputStream);
+ }
+
+ @Test
+ public void testOrderRead() throws Exception {
+ // write
+ int num = 10_000_000;
+ StreamStateHandle stateHandle = writeData(num);
+ Assert.assertNotNull(stateHandle);
+
+ int expectedStateSize = 2 * num * Integer.BYTES;
+ Assert.assertEquals(expectedStateSize, stateHandle.getStateSize());
+
+ // start read
+ FSDataInputStream fsDataInputStream = stateHandle.openInputStream();
+ FSDataBufferedInputStream bufferedInputStream = new FSDataBufferedInputStream(fsDataInputStream);
+ DataInputViewStreamWrapper inputView = new DataInputViewStreamWrapper(bufferedInputStream);
+
+ for (int i = 1; i <= num; i++) {
+ int read = inputView.readInt();
+ Assert.assertEquals(i, read);
+ }
+ for (int i = num; i >= 1; i--) {
+ int read = inputView.readInt();
+ Assert.assertEquals(i, read);
+ }
+ bufferedInputStream.close();
+ stateHandle.discardState();
+ }
+
+ @Test
+ public void testSeek() throws Exception {
Review comment:
@klion26 , thanks for your suggestion.
I have finished it. I hope you can help review again when you are free.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-724424008
> Thanks a lot for that deep diagnosis.
>
> Given that result, would it make sense to buffer all Hadoop streams in Flink (also affecting S3, etc.), or just the DFS streams (HDFS)? Do all streams have such high statistics/metrics costs, or just the HDFS input stream?
>
> In any case, it looks like we should add a `BufferedHadoopDataInputStream` that is a slight modification of the `HadoopDataInputStream` class with an internal buffer, and use that in the `HadoopFileSystem` when returning a stream. That way we should cover all cases.
It may be to make sense to buffer most of the streams in Flink.
First of all, it is beneficial to add buffer to `HadoopDataInputStream`. Secondly, there is no buffer in LocalDataInputStream, adding buffer can also improve IO performance. There will be some benefits for LocalRecovery.
For S3, I don't know much, and there is no S3 environment. I may not have a say.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-720452604
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a53411ad7bc59231419ea53bbcbfe3fb95ff67b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8794",
"triggerID" : "a53411ad7bc59231419ea53bbcbfe3fb95ff67b7",
"triggerType" : "PUSH"
}, {
"hash" : "91c08c8b38d94746603022eb07f85c3b4e3c095e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8835",
"triggerID" : "91c08c8b38d94746603022eb07f85c3b4e3c095e",
"triggerType" : "PUSH"
}, {
"hash" : "30162c89af86d6e2e666f9e119e98e7cf91548d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8839",
"triggerID" : "30162c89af86d6e2e666f9e119e98e7cf91548d3",
"triggerType" : "PUSH"
}, {
"hash" : "461553441634d634eb8950a60999a0eebc21b3dd",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8853",
"triggerID" : "461553441634d634eb8950a60999a0eebc21b3dd",
"triggerType" : "PUSH"
}, {
"hash" : "8872b60767d3e3ddf942564e1df3d2bb325fff6b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9086",
"triggerID" : "8872b60767d3e3ddf942564e1df3d2bb325fff6b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 461553441634d634eb8950a60999a0eebc21b3dd Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8853)
* 8872b60767d3e3ddf942564e1df3d2bb325fff6b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9086)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-731732373
Hi @StephanEwen
I've updated the PR based on your comments, Please help reviewing it when you have time.
Thanks.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-737999505
The PR looks to copy code from the Java SDK, which is Oracle licensed proprietary code, so this cannot be merged in any way.
@AHeise I think that is the reason for the unexpected formatting / visibility.
The PR also introduces again double wrapping: Hadoop's Stream -> `HadoopDataInputStream` -> `FSDataBufferedInputStream`
I would suggest to close this PR and go back to first defining what actually needs to be done, and how to approach this (not by copying copyrighted code!). Writing code here is the least important thing (introducing a buffer is not a complex problem) and it looks the PR takes a lot of iterations because it is always putting code in place before understanding how the problem should be approached.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-720452604
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a53411ad7bc59231419ea53bbcbfe3fb95ff67b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8794",
"triggerID" : "a53411ad7bc59231419ea53bbcbfe3fb95ff67b7",
"triggerType" : "PUSH"
}, {
"hash" : "91c08c8b38d94746603022eb07f85c3b4e3c095e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8835",
"triggerID" : "91c08c8b38d94746603022eb07f85c3b4e3c095e",
"triggerType" : "PUSH"
}, {
"hash" : "30162c89af86d6e2e666f9e119e98e7cf91548d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8839",
"triggerID" : "30162c89af86d6e2e666f9e119e98e7cf91548d3",
"triggerType" : "PUSH"
}, {
"hash" : "461553441634d634eb8950a60999a0eebc21b3dd",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8853",
"triggerID" : "461553441634d634eb8950a60999a0eebc21b3dd",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 461553441634d634eb8950a60999a0eebc21b3dd Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8853)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
AHeise commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-732319313
> `LocalFileSystem`
Yes, it looks like `LocalFileSystem` should also be a good candidate. In particular `#open(Path, int)` currently ignores the buffer size...
I'm wondering if it makes more sense to create two PRs though. Even though the solution is pretty much the same, the underlying issues are different imho. The smaller a PR, the faster we can usually go. However, since they are so related, I wouldn't mind both issues being resolved in the same PR.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-720452604
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a53411ad7bc59231419ea53bbcbfe3fb95ff67b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8794",
"triggerID" : "a53411ad7bc59231419ea53bbcbfe3fb95ff67b7",
"triggerType" : "PUSH"
}, {
"hash" : "91c08c8b38d94746603022eb07f85c3b4e3c095e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8835",
"triggerID" : "91c08c8b38d94746603022eb07f85c3b4e3c095e",
"triggerType" : "PUSH"
}, {
"hash" : "30162c89af86d6e2e666f9e119e98e7cf91548d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8839",
"triggerID" : "30162c89af86d6e2e666f9e119e98e7cf91548d3",
"triggerType" : "PUSH"
}, {
"hash" : "461553441634d634eb8950a60999a0eebc21b3dd",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8853",
"triggerID" : "461553441634d634eb8950a60999a0eebc21b3dd",
"triggerType" : "PUSH"
}, {
"hash" : "8872b60767d3e3ddf942564e1df3d2bb325fff6b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8872b60767d3e3ddf942564e1df3d2bb325fff6b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 461553441634d634eb8950a60999a0eebc21b3dd Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8853)
* 8872b60767d3e3ddf942564e1df3d2bb325fff6b UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-723377721
> Sorry to be late to the game here, but could you share a bit more information on what the original setup was?
> Specifically, what was your checkpoint storage system that offered such bad stream read performance? Was it HDFS? OSS? S3?
>
> Looking at this change here, it seems very big with more options and many changed classes, for "just" introducing a buffer in a stream. That makes me skeptical that this is fixed in the right place.
>
> Two other options to solve this:
>
> (1) Input stream buffering is a property of the `CheckpointStorage`. It is created there, rather than in the state backends that have to wrap the stream. That way it works for all users of the `CheckpointStorage`, not just the state backends that happened to be adjusted to wrap the stream.
>
> (2) Alternatively, we can make it a contract that all FileSystem implementations return well-buffered streams. Some already do this by default, wrapping them with another buffered stream adds just another layer and extra copying of bytes, costing performance. The ones that do not do that are easily adjusted.
>
> At a first glance, I'd say let's go with option (2) if possible, otherwise option (1).
> Hence the question: Which FS did you use that had such bad performing streams?
Thanks for your comments.
I use hdfs, mainly because there are a lot of small IO during restore. A Flink job is provided in jira to reproduce the test results.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-720452604
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a53411ad7bc59231419ea53bbcbfe3fb95ff67b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8794",
"triggerID" : "a53411ad7bc59231419ea53bbcbfe3fb95ff67b7",
"triggerType" : "PUSH"
}, {
"hash" : "91c08c8b38d94746603022eb07f85c3b4e3c095e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8835",
"triggerID" : "91c08c8b38d94746603022eb07f85c3b4e3c095e",
"triggerType" : "PUSH"
}, {
"hash" : "30162c89af86d6e2e666f9e119e98e7cf91548d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8839",
"triggerID" : "30162c89af86d6e2e666f9e119e98e7cf91548d3",
"triggerType" : "PUSH"
}, {
"hash" : "461553441634d634eb8950a60999a0eebc21b3dd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8853",
"triggerID" : "461553441634d634eb8950a60999a0eebc21b3dd",
"triggerType" : "PUSH"
}, {
"hash" : "8872b60767d3e3ddf942564e1df3d2bb325fff6b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9086",
"triggerID" : "8872b60767d3e3ddf942564e1df3d2bb325fff6b",
"triggerType" : "PUSH"
}, {
"hash" : "be7f3c7a5db0320f94ae49c57cb72f56c477bcdd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9902",
"triggerID" : "be7f3c7a5db0320f94ae49c57cb72f56c477bcdd",
"triggerType" : "PUSH"
}, {
"hash" : "4f972fcfd74e2d74c5cc7af7cc8561faadd21d1f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10395",
"triggerID" : "4f972fcfd74e2d74c5cc7af7cc8561faadd21d1f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4f972fcfd74e2d74c5cc7af7cc8561faadd21d1f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=10395)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-737892839
@AHeise
Thanks for your review. I am very busy at work recently, and may address your comment later.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-720452604
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a53411ad7bc59231419ea53bbcbfe3fb95ff67b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8794",
"triggerID" : "a53411ad7bc59231419ea53bbcbfe3fb95ff67b7",
"triggerType" : "PUSH"
}, {
"hash" : "91c08c8b38d94746603022eb07f85c3b4e3c095e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8835",
"triggerID" : "91c08c8b38d94746603022eb07f85c3b4e3c095e",
"triggerType" : "PUSH"
}, {
"hash" : "30162c89af86d6e2e666f9e119e98e7cf91548d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8839",
"triggerID" : "30162c89af86d6e2e666f9e119e98e7cf91548d3",
"triggerType" : "PUSH"
}, {
"hash" : "461553441634d634eb8950a60999a0eebc21b3dd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8853",
"triggerID" : "461553441634d634eb8950a60999a0eebc21b3dd",
"triggerType" : "PUSH"
}, {
"hash" : "8872b60767d3e3ddf942564e1df3d2bb325fff6b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9086",
"triggerID" : "8872b60767d3e3ddf942564e1df3d2bb325fff6b",
"triggerType" : "PUSH"
}, {
"hash" : "be7f3c7a5db0320f94ae49c57cb72f56c477bcdd",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "be7f3c7a5db0320f94ae49c57cb72f56c477bcdd",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8872b60767d3e3ddf942564e1df3d2bb325fff6b Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9086)
* be7f3c7a5db0320f94ae49c57cb72f56c477bcdd UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
Myasuka commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-734175097
As we use `OSSFileSystemFactory` for our production usage and it also wrapped as `HadoopFileSystem`, I think it's okay to introduce buffer in `HadoopFileSystem` and most DFS could benefit.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-724230980
Thanks a lot for that deep diagnosis.
Given that result, would it make sense to buffer all Hadoop streams in Flink (also affecting S3, etc.), or just the DFS streams (HDFS)? Do all streams have such high statistics/metrics costs, or just the HDFS input stream?
In any case, it looks like we should add a `BufferedHadoopDataInputStream` that is a slight modification of the `HadoopDataInputStream` class with an internal buffer, and use that in the `HadoopFileSystem` when returning a stream. That way we should cover all cases.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-720428549
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-724426185
I feel it can buffer all streams in Flink.
If `bufferSize> 0` then add buffer. If `bufferSize == 0`, the original InputStream is used. So if the buffer has a negative impact, the user can manually close the buffer. In theory, the buffer should not have a negative impact.
Please correct me if something I am missed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-731580036
Yes, I think that makes sense. Adding a `fs.buffer` config option, maybe with a default value of 64k?
- For local streams, it would be good as well, that is true.
- For S3, we get that implicitly if we buffer the `HadoopDataInputStream`, because the S3 reads go through some Hadoop utilities and also wrap the stream with the `HadoopDataInputStream` in the end.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-732264590
> Hi @1996fanrui ,
>
> that's an interesting discovery and investigation that you did there!
>
> I think the approach on the filesystem level is also much better than the previous way. Let's try to not change any public API (FileSystem) as this would slow down the progress.
>
> I'd probably focus on Hadoop file systems entirely (for now). What I'd propose is the following:
>
> * Use `HadoopFsFactory#configure` to extract the buffer size and pass it to the `ctor` of all filesystems created by the factory.
> * Use that default buffer size in `HadoopFileSystem#open(Path)` to call `#open(Path, int)`.
> * `HadoopFileSystem#open(Path, int)` should use the buffer size both in the call to Hadoop and to wrap it as you did as in the `BufferedFSInputStream`. I dug a bit into the Hadoop code and noticed that the cache is by default just 4kb. So even if we have cache on top of it with 64kb, we would still need to ask Hadoop several times.
>
> So, that means you are not adding any new methods, but just modify existing ones.
Hi @AHeise , thanks for your suggestions.
In the process of modifying the code, I also felt that the public API was modified. There may be some unknown effects.
I have some questions now: This feature can optimize the IO performance of `LocalFileSystem`, so can it be applied to `LocalFileSystem` and `HadoopFileSystem`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-731594944
> This should be totally independent from any checkpoint code, because it has nothing to do with checkpointing. This is purely file system business. I think we need to start a new implementation here, this PR is not in the right direction.
>
> The changes would be in the `FileSystem` class and the `FileSystemFactory` picks up the configuration value in the `configure()` method.
Ok, I will read `FileSystem` class. I will do my best to do it.
If I have any questions, I will ask you again.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-736582174
Hi @AHeise ,
I've updated the PR, Please help reviewing it when you have time.
Thanks.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on a change in pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on a change in pull request #13885:
URL: https://github.com/apache/flink/pull/13885#discussion_r518475754
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FSDataBufferedInputStream.java
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.state.filesystem;
+
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * FSDataInputStream with Buffer
+ */
+public class FSDataBufferedInputStream extends FSDataInputStream {
Review comment:
done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen edited a comment on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
StephanEwen edited a comment on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-731591635
This should be totally independent from any checkpoint code, because it has nothing to do with checkpointing. This is purely file system business. I think we need to start a new implementation here, this PR is not in the right direction.
The changes would be in the `FileSystem` class and the `FileSystemFactory` picks up the configuration value in the `configure()` method.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #13885:
URL: https://github.com/apache/flink/pull/13885#discussion_r534009700
##########
File path: flink-core/src/main/java/org/apache/flink/core/fs/FSDataBufferedInputStream.java
##########
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.fs;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * FSDataInputStream with Buffer, only support single thread.
+ * <p>
+ * The design of FSDataBufferedInputStream refers to {@link java.io.BufferedInputStream}.
+ * Compared to {@link java.io.BufferedInputStream}, the seek and getPos interfaces are mainly added.
+ */
+public class FSDataBufferedInputStream extends FSDataInputStream {
+
+ private static final int DEFAULT_BUFFER_SIZE = 8192;
+
+ protected byte[] buf;
+
+ // read offset of buf
+ private int pos;
+
+ // availed count of buf
+ private int count;
+
+ private final FSDataInputStream inputStream;
+
+ private boolean closed;
+
+ public FSDataBufferedInputStream(FSDataInputStream inputStream) {
+ this(inputStream, DEFAULT_BUFFER_SIZE);
+ }
+
Review comment:
This constructor is only used in tests, I'd remove it here and add the constant to the test.
##########
File path: flink-core/src/main/java/org/apache/flink/core/fs/FSDataBufferedInputStream.java
##########
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.fs;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * FSDataInputStream with Buffer, only support single thread.
+ * <p>
+ * The design of FSDataBufferedInputStream refers to {@link java.io.BufferedInputStream}.
+ * Compared to {@link java.io.BufferedInputStream}, the seek and getPos interfaces are mainly added.
+ */
+public class FSDataBufferedInputStream extends FSDataInputStream {
+
+ private static final int DEFAULT_BUFFER_SIZE = 8192;
+
+ protected byte[] buf;
+
+ // read offset of buf
+ private int pos;
+
+ // availed count of buf
+ private int count;
+
+ private final FSDataInputStream inputStream;
+
+ private boolean closed;
+
+ public FSDataBufferedInputStream(FSDataInputStream inputStream) {
+ this(inputStream, DEFAULT_BUFFER_SIZE);
+ }
+
+ public FSDataBufferedInputStream(
+ FSDataInputStream inputStream,
+ int bufferSize) {
Review comment:
nit: That should fit on one line.
##########
File path: flink-core/src/main/java/org/apache/flink/core/fs/FSDataBufferedInputStream.java
##########
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.fs;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * FSDataInputStream with Buffer, only support single thread.
+ * <p>
+ * The design of FSDataBufferedInputStream refers to {@link java.io.BufferedInputStream}.
+ * Compared to {@link java.io.BufferedInputStream}, the seek and getPos interfaces are mainly added.
+ */
+public class FSDataBufferedInputStream extends FSDataInputStream {
+
+ private static final int DEFAULT_BUFFER_SIZE = 8192;
+
+ protected byte[] buf;
+
+ // read offset of buf
+ private int pos;
+
+ // availed count of buf
+ private int count;
+
+ private final FSDataInputStream inputStream;
+
+ private boolean closed;
+
+ public FSDataBufferedInputStream(FSDataInputStream inputStream) {
+ this(inputStream, DEFAULT_BUFFER_SIZE);
+ }
+
+ public FSDataBufferedInputStream(
+ FSDataInputStream inputStream,
+ int bufferSize) {
+ this.inputStream = inputStream;
+
+ Preconditions.checkState(bufferSize > 0, "bufferSize must > 0");
+ this.buf = new byte[bufferSize];
+
+ this.pos = 0;
+ this.count = 0;
+ this.closed = false;
+ }
+
+ @Override
+ public void seek(long desired) throws IOException {
+ long streamPos = inputStream.getPos();
+ long bufStartPos = streamPos - count;
+ if (bufStartPos <= desired && desired < streamPos) {
+ this.pos = (int) (desired - bufStartPos);
+ return;
+ }
+ inputStream.seek(desired);
+ this.pos = 0;
+ this.count = 0;
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ int avail = count - pos;
+ return inputStream.getPos() - avail;
+ }
+
+ @Override
+ public int read() throws IOException {
+ if (pos >= count) {
+ fill();
+ if (pos >= count) {
+ return -1;
+ }
+ }
+ return buf[pos++] & 0xff;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ if ((off | len | (off + len) | (b.length - (off + len))) < 0) {
+ throw new IndexOutOfBoundsException();
+ } else if (len == 0) {
+ return 0;
+ }
+
+ int n = 0;
+ for (; ; ) {
+ int nread = read1(b, off + n, len - n);
+ if (nread <= 0) {
+ return (n == 0) ? nread : n;
+ }
+ n += nread;
+ if (n >= len) {
+ return n;
+ }
+ // if not closed but no bytes available, return
+ InputStream input = inputStream;
+ if (input != null && input.available() <= 0) {
+ return n;
+ }
+ }
+ }
+
+ /**
+ * Read characters into a portion of an array, reading from the underlying
+ * stream at most once if necessary.
+ */
+ private int read1(byte[] b, int off, int len) throws IOException {
+ int avail = count - pos;
+ if (avail <= 0) {
+ /* If the requested length is at least as large as the buffer,
+ do not bother to copy the bytes into the local buffer.
+ In this way buffered streams will cascade harmlessly. */
+ if (len >= buf.length) {
+ return inputStream.read(b, off, len);
+ }
+ fill();
+ avail = count - pos;
+ if (avail <= 0) {
+ return -1;
+ }
+ }
+ int cnt = Math.min(avail, len);
+ System.arraycopy(buf, pos, b, off, cnt);
+ pos += cnt;
+ return cnt;
+ }
+
+ @Override
+ public long skip(long n) throws IOException {
+ if (n <= 0) {
+ return 0;
+ }
+ long avail = count - pos;
+
+ if (avail <= 0) {
+ // Fill in buffer to save bytes for reset
+ fill();
+ avail = count - pos;
+ if (avail <= 0) {
+ return 0;
+ }
+ }
+
+ long skipped = Math.min(avail, n);
+ pos += skipped;
+ return skipped;
+ }
+
+ @Override
+ public int available() throws IOException {
+ int avail = count - pos;
+ return inputStream.available() + avail;
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ if (inputStream != null) {
+ inputStream.close();
+ }
+ }
Review comment:
I'd set `inputStream = null` after closing. Then you don't need `closed` at all - you could simply check if `inputStream != null`.
##########
File path: flink-filesystems/flink-hadoop-fs/src/test/java/org/apache/flink/runtime/fs/hdfs/HadoopFsFactoryTest.java
##########
@@ -61,4 +72,61 @@ public void testCreateHadoopFsWithMissingAuthority() throws Exception {
assertTrue(e.getMessage().contains("authority"));
}
}
+
+ @Test
+ public void testReadBufferSize() throws Exception {
+ final URI uri = URI.create("hdfs://localhost:12345/");
+ final Path path = new Path(uri.getPath());
+
+ org.apache.hadoop.fs.FileSystem hadoopFs = mock(org.apache.hadoop.fs.FileSystem.class);
+ org.apache.hadoop.fs.FSDataInputStream hadoopInputStream =
+ mock(org.apache.hadoop.fs.FSDataInputStream.class);
+ when(hadoopFs.open(isA(org.apache.hadoop.fs.Path.class), anyInt()))
+ .thenReturn(hadoopInputStream);
+
Review comment:
We actually want to move away from mocking with mockito as far as possible.
I'm wondering what's the benefit over using `file://` to some random (or temp) file?
##########
File path: flink-core/src/main/java/org/apache/flink/core/fs/FSDataBufferedInputStream.java
##########
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.fs;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * FSDataInputStream with Buffer, only support single thread.
+ * <p>
+ * The design of FSDataBufferedInputStream refers to {@link java.io.BufferedInputStream}.
+ * Compared to {@link java.io.BufferedInputStream}, the seek and getPos interfaces are mainly added.
+ */
+public class FSDataBufferedInputStream extends FSDataInputStream {
+
+ private static final int DEFAULT_BUFFER_SIZE = 8192;
+
+ protected byte[] buf;
Review comment:
Couldn't that be private?
##########
File path: flink-core/src/main/java/org/apache/flink/core/fs/FSDataBufferedInputStream.java
##########
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.fs;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * FSDataInputStream with Buffer, only support single thread.
+ * <p>
+ * The design of FSDataBufferedInputStream refers to {@link java.io.BufferedInputStream}.
+ * Compared to {@link java.io.BufferedInputStream}, the seek and getPos interfaces are mainly added.
+ */
+public class FSDataBufferedInputStream extends FSDataInputStream {
+
+ private static final int DEFAULT_BUFFER_SIZE = 8192;
+
+ protected byte[] buf;
+
+ // read offset of buf
+ private int pos;
+
+ // availed count of buf
+ private int count;
+
+ private final FSDataInputStream inputStream;
+
+ private boolean closed;
+
+ public FSDataBufferedInputStream(FSDataInputStream inputStream) {
+ this(inputStream, DEFAULT_BUFFER_SIZE);
+ }
+
+ public FSDataBufferedInputStream(
+ FSDataInputStream inputStream,
+ int bufferSize) {
+ this.inputStream = inputStream;
Review comment:
this.inputStream = Preconditions.checkNotNull(inputStream);
##########
File path: flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java
##########
@@ -116,17 +135,19 @@ public FileStatus getFileStatus(final Path f) throws IOException {
}
@Override
- public HadoopDataInputStream open(final Path f, final int bufferSize) throws IOException {
+ public FSDataInputStream open(final Path f, final int bufferSize) throws IOException {
final org.apache.hadoop.fs.Path path = toHadoopPath(f);
final org.apache.hadoop.fs.FSDataInputStream fdis = this.fs.open(path, bufferSize);
- return new HadoopDataInputStream(fdis);
+ HadoopDataInputStream inputStream = new HadoopDataInputStream(fdis);
+ if (readBufferSize <= 0) {
Review comment:
`readBufferSize == 0` is clearer -- it cannot be negative because of your ctor check. Also I'd add a constant for `0`. (`NO_BUFFER`, `NO_BUFFERING`, ...)
##########
File path: flink-core/src/main/java/org/apache/flink/core/fs/FSDataBufferedInputStream.java
##########
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.fs;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * FSDataInputStream with Buffer, only support single thread.
+ * <p>
+ * The design of FSDataBufferedInputStream refers to {@link java.io.BufferedInputStream}.
+ * Compared to {@link java.io.BufferedInputStream}, the seek and getPos interfaces are mainly added.
+ */
+public class FSDataBufferedInputStream extends FSDataInputStream {
+
+ private static final int DEFAULT_BUFFER_SIZE = 8192;
+
+ protected byte[] buf;
+
+ // read offset of buf
+ private int pos;
+
+ // availed count of buf
+ private int count;
+
+ private final FSDataInputStream inputStream;
+
+ private boolean closed;
+
+ public FSDataBufferedInputStream(FSDataInputStream inputStream) {
+ this(inputStream, DEFAULT_BUFFER_SIZE);
+ }
+
+ public FSDataBufferedInputStream(
+ FSDataInputStream inputStream,
+ int bufferSize) {
+ this.inputStream = inputStream;
+
+ Preconditions.checkState(bufferSize > 0, "bufferSize must > 0");
+ this.buf = new byte[bufferSize];
+
+ this.pos = 0;
+ this.count = 0;
+ this.closed = false;
+ }
+
+ @Override
+ public void seek(long desired) throws IOException {
+ long streamPos = inputStream.getPos();
+ long bufStartPos = streamPos - count;
+ if (bufStartPos <= desired && desired < streamPos) {
+ this.pos = (int) (desired - bufStartPos);
+ return;
+ }
+ inputStream.seek(desired);
+ this.pos = 0;
+ this.count = 0;
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ int avail = count - pos;
+ return inputStream.getPos() - avail;
+ }
+
+ @Override
+ public int read() throws IOException {
+ if (pos >= count) {
+ fill();
+ if (pos >= count) {
+ return -1;
+ }
+ }
+ return buf[pos++] & 0xff;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ if ((off | len | (off + len) | (b.length - (off + len))) < 0) {
+ throw new IndexOutOfBoundsException();
+ } else if (len == 0) {
+ return 0;
+ }
+
+ int n = 0;
+ for (; ; ) {
+ int nread = read1(b, off + n, len - n);
+ if (nread <= 0) {
+ return (n == 0) ? nread : n;
+ }
+ n += nread;
+ if (n >= len) {
+ return n;
+ }
+ // if not closed but no bytes available, return
+ InputStream input = inputStream;
+ if (input != null && input.available() <= 0) {
+ return n;
+ }
+ }
+ }
+
+ /**
+ * Read characters into a portion of an array, reading from the underlying
+ * stream at most once if necessary.
+ */
+ private int read1(byte[] b, int off, int len) throws IOException {
+ int avail = count - pos;
+ if (avail <= 0) {
+ /* If the requested length is at least as large as the buffer,
+ do not bother to copy the bytes into the local buffer.
+ In this way buffered streams will cascade harmlessly. */
+ if (len >= buf.length) {
+ return inputStream.read(b, off, len);
+ }
+ fill();
+ avail = count - pos;
+ if (avail <= 0) {
+ return -1;
+ }
+ }
+ int cnt = Math.min(avail, len);
+ System.arraycopy(buf, pos, b, off, cnt);
+ pos += cnt;
+ return cnt;
+ }
+
+ @Override
+ public long skip(long n) throws IOException {
+ if (n <= 0) {
+ return 0;
+ }
+ long avail = count - pos;
+
+ if (avail <= 0) {
+ // Fill in buffer to save bytes for reset
Review comment:
It feels as if we could improve `skip` by using `inputStream.skip`. Currently, `skip` is pretty much just `read` with discarding bytes. At least, S3InputStream and DFSInputStream have special implementations for skip.
##########
File path: flink-core/src/main/java/org/apache/flink/core/fs/FSDataBufferedInputStream.java
##########
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.fs;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * FSDataInputStream with Buffer, only support single thread.
+ * <p>
+ * The design of FSDataBufferedInputStream refers to {@link java.io.BufferedInputStream}.
+ * Compared to {@link java.io.BufferedInputStream}, the seek and getPos interfaces are mainly added.
+ */
+public class FSDataBufferedInputStream extends FSDataInputStream {
+
+ private static final int DEFAULT_BUFFER_SIZE = 8192;
+
+ protected byte[] buf;
+
+ // read offset of buf
+ private int pos;
+
+ // availed count of buf
+ private int count;
+
+ private final FSDataInputStream inputStream;
+
+ private boolean closed;
+
+ public FSDataBufferedInputStream(FSDataInputStream inputStream) {
+ this(inputStream, DEFAULT_BUFFER_SIZE);
+ }
+
+ public FSDataBufferedInputStream(
+ FSDataInputStream inputStream,
+ int bufferSize) {
+ this.inputStream = inputStream;
+
+ Preconditions.checkState(bufferSize > 0, "bufferSize must > 0");
+ this.buf = new byte[bufferSize];
+
+ this.pos = 0;
+ this.count = 0;
+ this.closed = false;
+ }
+
+ @Override
+ public void seek(long desired) throws IOException {
Review comment:
All operations should check if the stream has already been closed. I'd add `checkClosed();` as the first line to pretty much all methods.
See also my comment below - you don't need `closed`. `checkClosed could throw an IllegalStateException whenever, `inputStream == null`.
##########
File path: flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java
##########
@@ -51,7 +55,22 @@
* @param hadoopFileSystem The Hadoop FileSystem that will be used under the hood.
*/
public HadoopFileSystem(org.apache.hadoop.fs.FileSystem hadoopFileSystem) {
+ this(hadoopFileSystem, 0);
+ }
+
+ /**
+ * Wraps the given Hadoop File System object as a Flink File System object.
+ * The given Hadoop file system object is expected to be initialized already.
+ *
+ * @param hadoopFileSystem The Hadoop FileSystem that will be used under the hood.
+ * @param readBufferSize The size of the buffer to be used.
+ */
+ public HadoopFileSystem(org.apache.hadoop.fs.FileSystem hadoopFileSystem,
+ int readBufferSize) {
Review comment:
nit: formatting (either bring on one line, or chop down both parameters with double indent.
##########
File path: flink-core/src/main/java/org/apache/flink/core/fs/FSDataBufferedInputStream.java
##########
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.fs;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * FSDataInputStream with Buffer, only support single thread.
+ * <p>
+ * The design of FSDataBufferedInputStream refers to {@link java.io.BufferedInputStream}.
+ * Compared to {@link java.io.BufferedInputStream}, the seek and getPos interfaces are mainly added.
+ */
+public class FSDataBufferedInputStream extends FSDataInputStream {
+
+ private static final int DEFAULT_BUFFER_SIZE = 8192;
+
+ protected byte[] buf;
+
+ // read offset of buf
+ private int pos;
+
+ // availed count of buf
+ private int count;
+
+ private final FSDataInputStream inputStream;
+
+ private boolean closed;
+
+ public FSDataBufferedInputStream(FSDataInputStream inputStream) {
+ this(inputStream, DEFAULT_BUFFER_SIZE);
+ }
+
+ public FSDataBufferedInputStream(
+ FSDataInputStream inputStream,
+ int bufferSize) {
+ this.inputStream = inputStream;
+
+ Preconditions.checkState(bufferSize > 0, "bufferSize must > 0");
+ this.buf = new byte[bufferSize];
+
+ this.pos = 0;
+ this.count = 0;
+ this.closed = false;
+ }
+
+ @Override
+ public void seek(long desired) throws IOException {
+ long streamPos = inputStream.getPos();
+ long bufStartPos = streamPos - count;
+ if (bufStartPos <= desired && desired < streamPos) {
+ this.pos = (int) (desired - bufStartPos);
+ return;
+ }
+ inputStream.seek(desired);
+ this.pos = 0;
+ this.count = 0;
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ int avail = count - pos;
+ return inputStream.getPos() - avail;
+ }
+
+ @Override
+ public int read() throws IOException {
+ if (pos >= count) {
+ fill();
+ if (pos >= count) {
+ return -1;
+ }
+ }
+ return buf[pos++] & 0xff;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ if ((off | len | (off + len) | (b.length - (off + len))) < 0) {
+ throw new IndexOutOfBoundsException();
+ } else if (len == 0) {
+ return 0;
+ }
+
+ int n = 0;
+ for (; ; ) {
+ int nread = read1(b, off + n, len - n);
+ if (nread <= 0) {
+ return (n == 0) ? nread : n;
+ }
+ n += nread;
+ if (n >= len) {
+ return n;
+ }
+ // if not closed but no bytes available, return
+ InputStream input = inputStream;
+ if (input != null && input.available() <= 0) {
+ return n;
+ }
+ }
+ }
+
+ /**
+ * Read characters into a portion of an array, reading from the underlying
+ * stream at most once if necessary.
+ */
+ private int read1(byte[] b, int off, int len) throws IOException {
+ int avail = count - pos;
+ if (avail <= 0) {
+ /* If the requested length is at least as large as the buffer,
+ do not bother to copy the bytes into the local buffer.
+ In this way buffered streams will cascade harmlessly. */
+ if (len >= buf.length) {
+ return inputStream.read(b, off, len);
+ }
+ fill();
+ avail = count - pos;
+ if (avail <= 0) {
+ return -1;
+ }
+ }
+ int cnt = Math.min(avail, len);
+ System.arraycopy(buf, pos, b, off, cnt);
+ pos += cnt;
+ return cnt;
+ }
+
+ @Override
+ public long skip(long n) throws IOException {
+ if (n <= 0) {
+ return 0;
+ }
+ long avail = count - pos;
+
+ if (avail <= 0) {
+ // Fill in buffer to save bytes for reset
+ fill();
+ avail = count - pos;
+ if (avail <= 0) {
+ return 0;
+ }
+ }
+
+ long skipped = Math.min(avail, n);
+ pos += skipped;
+ return skipped;
+ }
+
+ @Override
+ public int available() throws IOException {
+ int avail = count - pos;
+ return inputStream.available() + avail;
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ if (inputStream != null) {
+ inputStream.close();
+ }
+ }
+
+ @VisibleForTesting
+ public int getBufferSize() {
+ return buf.length;
+ }
+
+ private void fill() throws IOException {
+ Preconditions.checkState(pos >= count);
+ count = inputStream.read(buf);
Review comment:
count can be `-1` here. Please double-check if it's used correctly in all places. `getPos` seems to not work for example. `seek` also looks suspicious.
##########
File path: flink-filesystems/flink-hadoop-fs/src/test/java/org/apache/flink/runtime/fs/hdfs/HadoopFsFactoryTest.java
##########
@@ -61,4 +72,61 @@ public void testCreateHadoopFsWithMissingAuthority() throws Exception {
assertTrue(e.getMessage().contains("authority"));
}
}
+
+ @Test
+ public void testReadBufferSize() throws Exception {
Review comment:
I'd probably divide the test into different cases: One without setting property, one with setting it to 0, one with setting it to >0. The additional lines should make the tests much easier to read.
##########
File path: flink-filesystems/flink-hadoop-fs/src/test/java/org/apache/flink/runtime/fs/hdfs/HadoopFsFactoryTest.java
##########
@@ -61,4 +72,61 @@ public void testCreateHadoopFsWithMissingAuthority() throws Exception {
assertTrue(e.getMessage().contains("authority"));
}
}
+
+ @Test
+ public void testReadBufferSize() throws Exception {
+ final URI uri = URI.create("hdfs://localhost:12345/");
+ final Path path = new Path(uri.getPath());
+
+ org.apache.hadoop.fs.FileSystem hadoopFs = mock(org.apache.hadoop.fs.FileSystem.class);
+ org.apache.hadoop.fs.FSDataInputStream hadoopInputStream =
+ mock(org.apache.hadoop.fs.FSDataInputStream.class);
+ when(hadoopFs.open(isA(org.apache.hadoop.fs.Path.class), anyInt()))
+ .thenReturn(hadoopInputStream);
+
+ // default configuration
+ Configuration configuration = new Configuration();
+
+ HadoopFsFactory fsFactory = new HadoopFsFactory();
+ fsFactory.configure(configuration);
+
+ FileSystem fileSystem = fsFactory.create(uri);
+ mockHadoopFsOpen(fileSystem, hadoopFs);
Review comment:
I hope that we don't need that as well...
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-720452604
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a53411ad7bc59231419ea53bbcbfe3fb95ff67b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8794",
"triggerID" : "a53411ad7bc59231419ea53bbcbfe3fb95ff67b7",
"triggerType" : "PUSH"
}, {
"hash" : "91c08c8b38d94746603022eb07f85c3b4e3c095e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8835",
"triggerID" : "91c08c8b38d94746603022eb07f85c3b4e3c095e",
"triggerType" : "PUSH"
}, {
"hash" : "30162c89af86d6e2e666f9e119e98e7cf91548d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8839",
"triggerID" : "30162c89af86d6e2e666f9e119e98e7cf91548d3",
"triggerType" : "PUSH"
}, {
"hash" : "461553441634d634eb8950a60999a0eebc21b3dd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8853",
"triggerID" : "461553441634d634eb8950a60999a0eebc21b3dd",
"triggerType" : "PUSH"
}, {
"hash" : "8872b60767d3e3ddf942564e1df3d2bb325fff6b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9086",
"triggerID" : "8872b60767d3e3ddf942564e1df3d2bb325fff6b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8872b60767d3e3ddf942564e1df3d2bb325fff6b Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9086)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-723193146
Sorry to be late to the game here, but could you share a bit more information on what the original setup was?
Specifically, what was your checkpoint storage system that offered such bad stream read performance? Was it HDFS? OSS? S3?
Looking at this change here, it seems very big with more options and many changed classes, for "just" introducing a buffer in a stream. That makes me skeptical that this is fixed in the right place.
Two other options to solve this:
(1) Input stream buffering is a property of the `CheckpointStorage`. It is created there, rather than in the state backends that have to wrap the stream. That way it works for all users of the `CheckpointStorage`, not just the state backends that happened to be adjusted to wrap the stream.
(2) Alternatively, we can make it a contract that all FileSystem implementations return well-buffered streams. Some already do this by default, wrapping them with another buffered stream adds just another layer and extra copying of bytes, costing performance. The ones that do not do that are easily adjusted.
At a first glance, I'd say let's go with option (2) if possible, otherwise option (1).
Hence the question: Which FS did you use that had such bad performing streams?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-720909799
@flinkbot run azure
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on a change in pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on a change in pull request #13885:
URL: https://github.com/apache/flink/pull/13885#discussion_r516392512
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
##########
@@ -524,13 +524,17 @@ public CheckpointStorageAccess createCheckpointStorage(JobID jobId) throws IOExc
HeapPriorityQueueSetFactory priorityQueueSetFactory =
new HeapPriorityQueueSetFactory(keyGroupRange, numberOfKeyGroups, 128);
+ ReadableConfig config = env.getTaskManagerInfo().getConfiguration();
+ int fsReadBufferSize = config.get(CheckpointingOptions.FS_READ_BUFFER_SIZE);
Review comment:
Hi, @sjwiesman , thanks for your comments. I will do it.
##########
File path: flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
##########
@@ -157,4 +157,14 @@
.withDescription(String.format("The default size of the write buffer for the checkpoint streams that write to file systems. " +
"The actual write buffer size is determined to be the maximum of the value of this option and option '%s'.", FS_SMALL_FILE_THRESHOLD.key()));
+ /**
+ * The default size of the read buffer for the checkpoint streams that read from file systems.
+ */
+ @Documentation.Section(Documentation.Sections.EXPERT_STATE_BACKENDS)
+ public static final ConfigOption<Integer> FS_READ_BUFFER_SIZE = ConfigOptions
+ .key("state.backend.fs.read-buffer-size")
+ .intType()
Review comment:
This configuration only means reading a buffer size of the file system. Should the `memory configuration` contain some `memory size` related config?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackend.java
##########
@@ -524,13 +524,17 @@ public CheckpointStorageAccess createCheckpointStorage(JobID jobId) throws IOExc
HeapPriorityQueueSetFactory priorityQueueSetFactory =
new HeapPriorityQueueSetFactory(keyGroupRange, numberOfKeyGroups, 128);
+ ReadableConfig config = env.getTaskManagerInfo().getConfiguration();
+ int fsReadBufferSize = config.get(CheckpointingOptions.FS_READ_BUFFER_SIZE);
Review comment:
done
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java
##########
@@ -331,13 +331,18 @@ public OperatorStateBackend createOperatorStateBackend(
TaskStateManager taskStateManager = env.getTaskStateManager();
HeapPriorityQueueSetFactory priorityQueueSetFactory =
new HeapPriorityQueueSetFactory(keyGroupRange, numberOfKeyGroups, 128);
+
+ ReadableConfig config = env.getTaskManagerInfo().getConfiguration();
+ int fsReadBufferSize = config.get(CheckpointingOptions.FS_READ_BUFFER_SIZE);
Review comment:
done
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockStateBackend.java
##########
@@ -134,13 +136,18 @@ public CheckpointStreamFactory resolveCheckpointStorageLocation(long checkpointI
MetricGroup metricGroup,
@Nonnull Collection<KeyedStateHandle> stateHandles,
CloseableRegistry cancelStreamRegistry) {
+
+ Configuration configuration = env.getTaskManagerInfo().getConfiguration();
+ int fsReadBufferSize = configuration.getInteger(CheckpointingOptions.FS_READ_BUFFER_SIZE);
Review comment:
done
##########
File path: flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
##########
@@ -540,6 +540,10 @@ public CheckpointStorageAccess createCheckpointStorage(JobID jobId) throws IOExc
ExecutionConfig executionConfig = env.getExecutionConfig();
StreamCompressionDecorator keyGroupCompressionDecorator = getCompressionDecorator(executionConfig);
+
+ ReadableConfig config = env.getTaskManagerInfo().getConfiguration();
+ int fsReadBufferSize = config.get(CheckpointingOptions.FS_READ_BUFFER_SIZE);
Review comment:
done
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FSDataBufferedInputStreamTest.java
##########
@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.state.filesystem;
+
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.runtime.state.StreamStateHandle;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.UUID;
+
+/**
+ * Unit tests for the {@link FSDataBufferedInputStream}.
+ */
+public class FSDataBufferedInputStreamTest {
+
+ private static final int KB = 1024;
+ private static final int MB = 1024 * KB;
+
+ FsCheckpointStreamFactory.FsCheckpointStateOutputStream outputStream;
+ DataOutputViewStreamWrapper outView;
+
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Before
+ public void before() throws IOException {
+ String path = temporaryFolder.newFolder().getAbsolutePath();
+ File outFile = new File(path, String.valueOf(UUID.randomUUID()));
+
+ Path outPath = new Path(outFile.toURI());
+ outputStream = new FsCheckpointStreamFactory.FsCheckpointStateOutputStream(
+ outPath, outPath.getFileSystem(), MB, KB);
+ outView = new DataOutputViewStreamWrapper(outputStream);
+ }
+
+ @Test
+ public void testOrderRead() throws Exception {
+ // write
+ int num = 10_000_000;
+ StreamStateHandle stateHandle = writeData(num);
+ Assert.assertNotNull(stateHandle);
+
+ int expectedStateSize = 2 * num * Integer.BYTES;
+ Assert.assertEquals(expectedStateSize, stateHandle.getStateSize());
+
+ // start read
+ FSDataInputStream fsDataInputStream = stateHandle.openInputStream();
+ FSDataBufferedInputStream bufferedInputStream = new FSDataBufferedInputStream(fsDataInputStream);
+ DataInputViewStreamWrapper inputView = new DataInputViewStreamWrapper(bufferedInputStream);
+
+ for (int i = 1; i <= num; i++) {
+ int read = inputView.readInt();
+ Assert.assertEquals(i, read);
+ }
+ for (int i = num; i >= 1; i--) {
+ int read = inputView.readInt();
+ Assert.assertEquals(i, read);
+ }
+ bufferedInputStream.close();
+ stateHandle.discardState();
+ }
+
+ @Test
+ public void testSeek() throws Exception {
Review comment:
@klion26 , thanks for your suggest.
I have finished it. I hope you can help review again when you are free.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
1996fanrui commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-741595105
@StephanEwen
Sorry, because I ignored the copyright. I will close this PR.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-731591635
This should be totally independent from any checkpoint code, because it has nothing to do with checkpointing. This is purely file system business.
The changes would be in the `FileSystem` class and the `FileSystemFactory` picks up the configuration value in the `configure()` method.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-720452604
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a53411ad7bc59231419ea53bbcbfe3fb95ff67b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8794",
"triggerID" : "a53411ad7bc59231419ea53bbcbfe3fb95ff67b7",
"triggerType" : "PUSH"
}, {
"hash" : "91c08c8b38d94746603022eb07f85c3b4e3c095e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8835",
"triggerID" : "91c08c8b38d94746603022eb07f85c3b4e3c095e",
"triggerType" : "PUSH"
}, {
"hash" : "30162c89af86d6e2e666f9e119e98e7cf91548d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8839",
"triggerID" : "30162c89af86d6e2e666f9e119e98e7cf91548d3",
"triggerType" : "PUSH"
}, {
"hash" : "461553441634d634eb8950a60999a0eebc21b3dd",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8853",
"triggerID" : "461553441634d634eb8950a60999a0eebc21b3dd",
"triggerType" : "PUSH"
}, {
"hash" : "8872b60767d3e3ddf942564e1df3d2bb325fff6b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9086",
"triggerID" : "8872b60767d3e3ddf942564e1df3d2bb325fff6b",
"triggerType" : "PUSH"
}, {
"hash" : "be7f3c7a5db0320f94ae49c57cb72f56c477bcdd",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9902",
"triggerID" : "be7f3c7a5db0320f94ae49c57cb72f56c477bcdd",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* be7f3c7a5db0320f94ae49c57cb72f56c477bcdd Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=9902)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on pull request #13885: [FLINK-19911] add read buffer for input stream
Posted by GitBox <gi...@apache.org>.
AHeise commented on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-736035288
> Based on the above design, there are some questions that I hope to be answered:
>
> * `buffer size` is passed to HadoopFileSystem through Constructor. It means that HadoopFileSystem needs to add a new Constructor: `HadoopFileSystem(FileSystem, bufferSize)`. HadoopFsFactory will call the new constructor.
>
> * The old constructor `HadoopFileSystem(FileSystem)` is still called by other FsFactory. So other FsFactory will not be able to improve performance. For example: OSSFileSystemFactory.
>
>
> Question: Do other FsFactory related to HadoopFileSystem need to improve performance? If needed, are there other better designs. (Can there be a way not to modify many FsFactory?)
>
> Thanks.
Similar to your initial design, I would make `bufferSize = 0` => no buffering. Then all other filesystem factories could use the changed constructor with `HadoopFileSystem(FileSystem, NO_BUFFER)` (`NO_BUFFER = 0`).
Since this PR is very specific to Hadoop, I'd probably set the buffer only through `HadoopFsFactory`. Your finding would not necessarily also transfer to other implementations of the `hadoop.FileSystem` (i.e., they don't have statistics). We could evaluate that in future PRs.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui edited a comment on pull request #13885: [FLINK-19911] Read checkpoint stream with buffer to speedup restore
Posted by GitBox <gi...@apache.org>.
1996fanrui edited a comment on pull request #13885:
URL: https://github.com/apache/flink/pull/13885#issuecomment-724092762
> Are the HDFS input streams generally not buffered? Would it make sense to adjust the `HadoopDataInputStream` class to be buffered?
Hi @StephanEwen , I am sorry to reply you so late.
In the past two days, I read the relevant code of hdfs reading data and did some performance analysis.
conclusion as below:
The hdfs input stream has buffer by default. The default buffer size is 64KB.
In theory, Use FSDataBufferedInputStream to wrap hdfsInputStream does not reduce the number of disk accesses. But the result shows: When buffer is added, restore time is reduced to one-third of the original.
So, I did some performance analysis.
From the point of view of CPU usage, When buffer is added cpu usage is greatly reduced.
Using hdfsInputStream directly, the CPU usage rate is 60-70%.
Use FSDataBufferedInputStream to wrap hdfsInputStream, the CPU usage rate is 20-25%.
Analyze why hdfsInputStream consumes CPU:
The hdfs client contains a lot of statistical information, and the method call stack is relatively deep, each method will consume a little performance. As shown in the picture below:
[flame graph remark link](https://drive.google.com/file/d/1zTwHdmSybAgyBGIIP71FLfMQ5DvK1eON/view?usp=sharing)
If you use FSDataBufferedInputStream to wrap hdfsInputStream, there will be a buffer outside the hdfs client, avoiding a very deep call stack and a lot of hdfs statistics.
Original flame graph link:
[restore with buffer](https://drive.google.com/file/d/1jxBNzh2iIsrX__wfFjvTWiC9LPCCDS3A/view?usp=sharing)
[restore without buffer](https://drive.google.com/file/d/1qDbqQC4bG34_ZsCrOpDaNouNQpvVXRWH/view?usp=sharing)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org