[
https://issues.apache.org/jira/browse/FLINK-10653?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
zhijiang updated FLINK-10653:
-----------------------------
Description:
This is the umbrella issue for improving shuffle architecture.
Shuffle is the process of data transfer between stages, which involves in
writing outputs on sender side and reading data on receiver side. In flink
implementation, it covers three parts of writer, transport layer and reader
separately which are uniformed for both streaming and batch jobs.
In detail, the current ResultPartitionWriter interface on upstream side only
supports in-memory outputs for streaming job and local persistent file outputs
for batch job. If we extend to implement another writer such as DfsWriter,
RdmaWriter, SortMergeWriter, etc based on ResultPartitionWriter interface, it
has not the unified mechanism to extend the reader side accordingly.
In order to make the shuffle architecture more flexible and support more
scenarios especially for batch jobs, a high level shuffle architecture is
necessary to manage and extend both writer and reader sides together.
Refer to the design doc for more details.
was:
This is the umbrella issue for improving shuffle for batch jobs.
The shuffle behavior in flink can be improved in two dimensions:
*1. ResultPartition:* There are two types of subpartitions for the sender’s
outputs. The PipelinedSubPartition is used for stream jobs and the
SpillableSubpartition is used for batch jobs in blocking mode.
* *Single file mode*: The spillable subpartition generates one separate
persistent file for each subpartition. This partition-hash mode is not IO
friendly in some scenarios, i.e. large number of subpartitions with little data
in each subpartition. Even it may exceed the inode limits for large scale jobs.
* *Failover case*: The persistent files would be deleted immediately after
reading to transport layer. If the downstream task fails during consumption,
all the related upstream tasks have to be restarted to reproduce the outputs.
*2. ShuffleService:* Shuffle service is used for transporting sender’s data to
the receiver side. In flink, TaskExecutor takes the role of shuffle service
unified for both stream and batch jobs.
* *Resource utility*: The output is consumable only when the sender’s task
finishes for blocking mode. TaskExecutor can not exit to release resources even
though there are no running tasks, because it has to serve for transferring
data via shuffle service. This behavior has the bad effects in resource dynamic
or sensitive scenarios.
* *IO performance*: Many TaskExecutors deployed on one node may touch disk IO
frequently for shuffle service at the same time. Loss of global control may
decrease IO performance in sensitive scenarios.
We propose three overall changes:
* Define the pluggable ShuffleManager interface
* Mechanism of external shuffle service
* Clear persistent files via notification or TTL
Refer to the design doc for more details.
> Fine-grained Shuffle System
> ---------------------------
>
> Key: FLINK-10653
> URL: https://issues.apache.org/jira/browse/FLINK-10653
> Project: Flink
> Issue Type: New Feature
> Components: Network
> Affects Versions: 2.0.0
> Reporter: zhijiang
> Assignee: zhijiang
> Priority: Major
>
> This is the umbrella issue for improving shuffle architecture.
> Shuffle is the process of data transfer between stages, which involves in
> writing outputs on sender side and reading data on receiver side. In flink
> implementation, it covers three parts of writer, transport layer and reader
> separately which are uniformed for both streaming and batch jobs.
> In detail, the current ResultPartitionWriter interface on upstream side only
> supports in-memory outputs for streaming job and local persistent file
> outputs for batch job. If we extend to implement another writer such as
> DfsWriter, RdmaWriter, SortMergeWriter, etc based on ResultPartitionWriter
> interface, it has not the unified mechanism to extend the reader side
> accordingly.
> In order to make the shuffle architecture more flexible and support more
> scenarios especially for batch jobs, a high level shuffle architecture is
> necessary to manage and extend both writer and reader sides together.
> Refer to the design doc for more details.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
