[jira] [Commented] (FLINK-26803) Merge small ChannelState file for Unaligned Checkpoint

["Anton Kalashnikov (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-26803?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17522261#comment-17522261 ] 

Anton Kalashnikov commented on FLINK-26803:
-------------------------------------------

[~fanrui], thanks for this idea. I did some research about this(extra thanks to [~roman] for feedback and explanation). Some results:
 Every parallel instance(operator chain) creates:
* one file for input + output channel state
* one file for each operator in the chain
So `number of files = number of chains * parallelism + state backend * parallelism`

As I understand, the main reason why Flink has such logic is the simplicity of implementation + easy rescaling. Theoretically, the problem with rescaling can be resolved by adding the offset with a specific subtaskIndex but there is still the problem with creating `stream factory` on `TaskManager` since Flink don't know in advance all task that will be executed on it. Also reading should be changed as well to avoid loading one big file multiply times.

You also can check this configuration [state.storage.fs.memory-threshold|https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/#state-storage-fs-memory-threshold]. It helps to reduce the number of the files if states are small enough. Maybe it will help you.

In conclusion, I don't really sure that it makes sense to implement this feature since it looks too complicated but the advantages are not so obvious. But of course I can be wrong somewhere and we can discuss it. But anyway, I think we need firstly to collect all advantages and try to understand possible implementation.

> Merge small ChannelState file for Unaligned Checkpoint
> ------------------------------------------------------
>
>                 Key: FLINK-26803
>                 URL: https://issues.apache.org/jira/browse/FLINK-26803
>             Project: Flink
>          Issue Type: Improvement
>          Components: Runtime / Checkpointing, Runtime / Network
>            Reporter: fanrui
>            Priority: Major
>
> When making an unaligned checkpoint, the number of ChannelState files is TaskNumber * subtaskNumber. For high parallelism job, it writes too many small files. It causes high load for hdfs NN.
>  
> In our production, a job writes more than 50K small files for each Unaligned Checkpoint. Could we merge these files before write FileSystem? We can configure the maximum number of files each TM can write in a single Unaligned Checkpoint.



--
This message was sent by Atlassian Jira
(v8.20.1#820001)