[
https://issues.apache.org/jira/browse/FLINK-33734?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17794682#comment-17794682
]
Roman Khachatryan commented on FLINK-33734:
-------------------------------------------
Thanks for the proposal!
You're right [~Feifan Wang], we do need to add a new Metadata Serializer
version.
Actually, I think it should be possible to solve the original problem (big
_metadata file) by only changing how these state handles are (de)serialized.
I.e. merge them in JM in MetadataSerializer. That would be less efficient of
course (as it wouldn't reduce RPC message size), but the code might be simpler
and the change easier to implement. To be clear, I personally don't have any
preference.
As for recovery, it's not only about performance, but also RPC message size:
with checkpointing improved (but not recovery), RPC message size might be
significantly larger on recovery than on checkpoint. We should probably
document this (and add a feature toggle), until recovery is also optimized.
I have a couple of more questions:
1. Does it make sense to also merge state from multiple subtasks (as
implemented in [FLINK-26803|https://issues.apache.org/jira/browse/FLINK-26803])?
2. What happens when the delegate is in-memory state handle
(`ByteStreamStateHandle`)?
> Merge unaligned checkpoint state handle
> ---------------------------------------
>
> Key: FLINK-33734
> URL: https://issues.apache.org/jira/browse/FLINK-33734
> Project: Flink
> Issue Type: Improvement
> Components: Runtime / Checkpointing
> Reporter: Feifan Wang
> Assignee: Feifan Wang
> Priority: Major
>
> h3. Background
> Unaligned checkpoint will write the inflight-data of all InputChannel and
> ResultSubpartition of the same subtask to the same file during checkpoint.
> The InputChannelStateHandle and ResultSubpartitionStateHandle organize the
> metadata of inflight-data at the channel granularity, which causes the file
> name to be repeated many times. When a job is under backpressure and task
> parallelism is high, the metadata of unaligned checkpoints will bloat. This
> will result in:
> # The amount of data reported by taskmanager to jobmanager increases, and
> jobmanager takes longer to process these RPC requests.
> # The metadata of the entire checkpoint becomes very large, and it takes
> longer to serialize and write it to dfs.
> Both of the above points ultimately lead to longer checkpoint duration.
> h3. A Production example
> Take our production job with a parallelism of 4800 as an example:
> # When there is no back pressure, checkpoint end-to-end duration is within 7
> seconds.
> # When under pressure: checkpoint end-to-end duration often exceeds 1
> minute. We found that jobmanager took more than 40 seconds to process rpc
> requests, and serialized metadata took more than 20 seconds.Some checkpoint
> statistics:
> |metadata file size|950 MB|
> |channel state count|12,229,854|
> |channel file count|5536|
> Of the 950MB in the metadata file, 68% are redundant file paths.
> We enabled log-based checkpoint on this job and hoped that the checkpoint
> could be completed within 30 seconds. This problem made it difficult to
> achieve this goal.
> h3. Propose changes
> I suggest introducing MergedInputChannelStateHandle and
> MergedResultSubpartitionStateHandle to eliminate redundant file paths.
> The taskmanager merges all InputChannelStateHandles with the same delegated
> StreamStateHandle in the same subtask into one MergedInputChannelStateHandle
> before reporting. When recovering from checkpoint, jobmangager converts
> MergedInputChannelStateHandle to InputChannelStateHandle collection before
> assigning state handle, and the rest of the process does not need to be
> changed.
> Structure of MergedInputChannelStateHandle :
>
> {code:java}
> { // MergedInputChannelStateHandle
> "delegate": {
> "filePath":
> "viewfs://hadoop-meituan/flink-yg15/checkpoints/retained/1234567/ab8d0c2f02a47586490b15e7a2c30555/chk-31/ffe54c0a-9b6e-4724-aae7-61b96bf8b1cf",
> "stateSize": 123456
> },
> "size": 2000,
> "subtaskIndex":0,
> "channels": [ // One InputChannel per element
> {
> "info": {
> "gateIdx": 0,
> "inputChannelIdx": 0
> },
> "offsets": [
> 100,200,300,400
> ],
> "size": 1400
> },
> {
> "info": {
> "gateIdx": 0,
> "inputChannelIdx": 1
> },
> "offsets": [
> 500,600
> ],
> "size": 600
> }
> ]
> }
> {code}
> MergedResultSubpartitionStateHandle is similar.
>
>
> WDYT [~roman] , [~pnowojski] , [~fanrui] ?
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
