[FLINK-19911] add read buffer for input stream

• Type: Improvement
• Status: Closed
• Priority: Not a Priority
• Resolution: Duplicate
• Affects Version/s: 1.11.3, 1.12.0, 1.13.0
• Component/s: FileSystems, Runtime / Checkpointing, Runtime / State Backends
• Labels: auto-deprioritized-major auto-deprioritized-minor auto-unassigned pull-request-available

Environment

Flink version: 1.10
StateBackend : FsStateBackend
code: Flink SQL count(distinct userId)
uv: 10 million
State size: 200M
TM total memory: 16G
Parallelism: 1

• Assignee: Unassigned
• Reporter: Rui Fan
• Created: 01/Nov/20 13:10
• Updated: 25/Sep/23 03:49
• Resolved: 21/Aug/23 09:07

Description

Heap StateBackend needs to serialize each Java Object into the file system during snapshot. RocksDB StateBackend's RocksFullSnapshotStrategy needs to read kvs from RocksDB and write them to the file system in the snapshot.

The above two cases involve a lot of small io, not large io, frequent small io is not friendly to disk. Therefore, the buffer is used in the checkpoint snapshot writing process of the file system. For details, refer to the buffer of FsCheckpointStreamFactory.FsCheckpointStateOutputStream.

There will be many small IOs in the restore process, but restore does not have a buffer. So I added a buffer and tested it based on Flink job.

Flink Job environment:

 Flink version: 1.10
 StateBackend : FsStateBackend 
 code: Flink SQL count(distinct userId)
 uv: 10 million
 State size: 200M
 TM total memory: 16G
 Parallelism: 1

It takes 33.1s to restore without read buffer, and 12.8s to restore with read buffer.

How to do it？

Use FSDataBufferedInputStream to wrap fsDataInputStream in HeapRestoreOperation#restore，code：

FSDataInputStream fsDataInputStream = keyGroupsStateHandle.openInputStream();
FSDataInputStream bufferedInputStream = new FSDataBufferedInputStream(fsDataInputStream);

 

Issue Links

• is duplicated by: FLINK-26586 FileSystem uses unbuffered read I/O
• is duplicated by: FLINK-33127 HeapKeyedStateBackend: use buffered I/O to speed up local recovery
• links to: GitHub Pull Request #13885