[FLINK-10653] Introduce Pluggable Shuffle Service Architecture

• Type: New Feature
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.9.0
• Component/s: Runtime / Network
• Assignee: Zhijiang
• Reporter: Zhijiang
• Created: 23/Oct/18 10:12
• Updated: 09/Oct/20 07:07
• Resolved: 09/Oct/20 07:04

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.

Issue Links

• is a parent of: FLINK-13247 Implement external shuffle service for YARN
• is a parent of: FLINK-11805 A Common External Shuffle Service Framework
• is a parent of: FLINK-13246 Implement external shuffle service for Kubernetes
• relates to: FLINK-1833 Refactor partition availability notification in ExecutionGraph
• supercedes: FLINK-1833 Refactor partition availability notification in ExecutionGraph
• links to: FLIP-31

Sub-Tasks

• Introduce ResultPartitionWithConsumableNotifier in task for notifying consumable result partition: Resolved
• Remove the schedule mode property from RPDD to TDD: Closed
• Introduce ShuffleMaster in Job Master: Resolved
• Introduce ShuffleEnvironment in Task Executor: Resolved
• Activate default shuffle implementation and remove legacy code: Closed
• Extend the necessary methods in ResultPartitionWriter interface: Closed
• Make ResultPartitionWriter extend AutoCloseable: Closed
• Make InputGate interface extend AutoCloseable: Closed
• Remove IOMode from NetworkEnvironment: Closed
• Remove KvState related components from NetworkEnvironment: Closed
• Refactor the creation of ResultPartition and InputGate into NetworkEnvironment: Closed
• Replace IntermediateResultPartitionID with ResultPartitionID in ResultPartitionDeploymentDescriptor: Closed
• Refactor to simplify the process of scheduleOrUpdateConsumers: Resolved
• Refactor the constructor of NetworkEnvironment: Closed
• Abstract TaskEventPublisher interface for simplifying NetworkEnvironment: Closed
• Move network related options to NetworkEnvironmentOptions: Closed
• Remove legacy fields for SingleInputGate: Closed
• Remove unregister task from NetworkEnvironment to simplify the interface of ShuffleService: Closed
• Introduce partition/gate setup to decouple task registration with NetworkEnvironment: Closed
• Refactor IOMetrics to not distinguish between local/remote in/out bytes: Closed
• Introduce InputGateWithMetrics in Task to increment numBytesIn metric: Closed
• Consider introducing batch metric register in NetworkEnviroment: Closed
• Refactor ResultPartitionManager to break tie with Task: Resolved
• Move network metrics setup into NetworkEnvironment: Closed
• Refactor the start method of ConnectionManager: Closed
• Move Task.inputGatesById to NetworkEnvironment: Closed
• Introduce an encapsulated metric group layout for shuffle API and deprecate old one: Closed
• Remove getBufferProvider method from ResultPartitionWriter interface: Closed
• Switch Task from ResultPartition to ResultPartitionWriter interface: Closed
• Make NetworkEnvironment#start() return the binded data port: Closed
• Remove getOwningTaskName method from InputGate: Resolved
• Refactor abstract InputGate to general interface: Closed
• Introduce NetworkEnvironment.getUnreleasedPartitions instead of using getResultPartitionManager: Closed
• Introduce ShuffleService interface and its configuration: Resolved
• Make shuffle environment implementation independent with IOManager: Closed
• Remove abstract getPageSize method from InputGate: Resolved
• Remove ExecutionAttemptID argument from ResultPartitionFactory#create: Closed
• Add partition lifecycle related Shuffle API: Closed
• Introduce ShuffleDescriptor#ReleaseType and ShuffleDescriptor#getSupportedReleaseTypes: Closed
• Refactor the process of SchedulerNG#requestPartitionState: Closed
• Remove getBufferSize method from BufferPoolFactory: Closed
• Remove ShuffleDescriptor.ReleaseType and make release semantics fixed per partition type: Closed