Shawnsuun opened a new pull request, #25838:
URL: https://github.com/apache/flink/pull/25838
<!--
*Thank you very much for contributing to Apache Flink - we are happy that
you want to help us improve Flink. To help the community review your
contribution in the best possible way, please go through the checklist below,
which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a
hassle. In order to uphold a high standard of quality for code contributions,
while at the same time managing a large number of contributions, we need
contributors to prepare the contributions well, and give reviewers enough
contextual information for the review. Please also understand that
contributions that do not follow this guide will take longer to review and thus
typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA
issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are
made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the
pull request", where *FLINK-XXXX* should be replaced by the actual issue
number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following
this pattern: `[hotfix] [docs] Fix typo in event time introduction` or
`[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the
pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean
verify` passes. You can set up Azure Pipelines CI to do that following [this
guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from
multiple issues.
- Each commit in the pull request has a meaningful commit message
(including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and
this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
Currently, when a Flink job finishes, it writes an archive as a single file
that maps paths to JSON files. Flink History Server (FHS) job archives are
pulled locally to where the FHS is running. This process creates a local
directory structure that scales inefficiently as the number of jobs increases.
### Key Problems
- **High inode usage** in the file system due to nested directories for job
archives.
- **Slower data retrieval** and bottlenecks in job archive navigation at
scale.
- Challenges due to limited file system scalability.
### Proposed Solution
Integrating **RocksDB**, a high-performance embedded database, as an
alternative storage backend for job archives. RocksDB provides:
- **Faster job data retrieval.**
- **Reduced inode consumption.**
- **Enhanced scalability**, especially in containerized environments.
The integration of RocksDB is implemented as a pluggable backend. The
current file system storage remains intact, while RocksDB serves as an optional
alternative for efficient storage and retrieval of job archives.
---
## Brief Change Log
### 1. KVStore Interface
- Introduced `KVStore` as an abstraction for key-value storage systems to
enable flexible storage backends.
- Added basic CRUD operations and advanced capabilities for managing job
archives.
### 2. RocksDB Integration
- Implemented `HistoryServerRocksDBKVStore` as the RocksDB-based
implementation of the `KVStore` interface.
- Mapped the hierarchical file-based job archive structure into key-value
pairs for efficient storage and retrieval.
### 3. ArchiveFetcher Abstraction and Improvements
- Introduced `ArchiveFetcher` as an abstract class to support multiple
backends for job archive fetching.
- Updated `HistoryServerArchiveFetcher` for file-based systems.
- Created `HistoryServerKVStoreArchiveFetcher` to fetch job archives using
RocksDB.
### 4. ServerHandler Abstraction and Improvements
- Designed `HistoryServerServerHandler` as an abstract base class for
handling HTTP requests, supporting pluggable backends.
- Updated `HistoryServerStaticFileServerHandler` for file-based job archive
serving.
- Implemented `HistoryServerKVStoreServerHandler` to serve job data from
RocksDB via REST APIs.
### 5. HistoryServer Updates
- Modified `HistoryServer` to integrate the `KVStore` interface and support
RocksDB as a pluggable backend.
- Added configuration options in `HistoryServerOptions` to toggle between
file-based and RocksDB storagen:
- Add the following configuration options in your flink-conf.yaml file to
enable RocksDB as the storage backend for the History Server.
```yaml
historyserver.storage.backend: kvstore
```
---
## Verifying this change
This change added tests and can be verified as follows:
### 1. Testing
- **Unit Tests**:
- Added `FhsRocksDBKVStoreTest` to validate CRUD operations and resource
cleanup for RocksDB.
- Added `HistoryServerKVStoreArchiveFetcherTest` to ensure correct
fetching and processing of job archives from RocksDB.
- **Integration Tests**:
- Built a Flink binary and configured `flink-conf.yaml` to test both
file-based and RocksDB backends.
- Verified archive retrieval via the History Server web UI and ensured
backward compatibility with the file-based backend.
- **End-to-End Tests**:
- Conducted tests in a Kubernetes cluster with both RocksDB and file-based
storage backends.
- Verified correct behavior of the History Server in processing and
displaying job archives for both storage backends in a real-world setup.
### 2. Performance Enhancements
- **Faster Archive Retrieval**: Achieved a 4.25x improvement in fetching and
processing archives with RocksDB compared to the traditional file system
(tested in a production environment).
- File system: 17 minutes for 100 archives.
- RocksDB: 4 minutes for 100 archives.
- **Reduced Inode Usage**: Reduced inode consumption by over 99.99%.
- File system: Over 20 million inodes.
- RocksDB: Only 79 inodes.
- **Lower Storage Usage**: Achieved a 95.6% reduction in storage usage.
- File system: 48 GB for 100 archives.
- RocksDB: 2.1 GB for 100 archives.
These enhancements significantly improve scalability, reduce resource
overhead, and make the History Server more responsive for large-scale
deployments.
---
## Does this pull request potentially affect one of the following parts:
- **Dependencies**: No (using existing RocksDB dependency).
- **Public API**: No.
- **Serializers**: No.
- **Performance-sensitive code paths**: Yes (job archive storage and
retrieval).
- **Deployment or recovery**: Yes (affects FHS deployment with the RocksDB
backend option).
- **File system connectors**: No.
---
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (not documented)
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
