...
We intend to introduce the following new configuration parameters.
Key | Type | Default Value | Description |
execution.batch.job-recovery.enabled | Boolean | false | A flag to enable or disable the job recovery. |
execution.batch.job-recovery.previous-worker.recovery.timeout | Duration | 30 s | The timeout for a new job master to wait for the previous worker to reconnect. |
execution.batch.job-recovery.operator-coordinator-snapshot.min-pause | Duration | 3 min | The minimal pause between operator coordinator snapshots. |
job-event.store.write-buffer.size | MemorySize | 1M | The size of the write buffer of JobEventStore, the content will be flushed to external file system once it's full |
job-event.store.write-buffer.flush-interval | Duration | 1 s | The flush interval of write buffer of JobEventStore, over this time, the content will be flushed to external file system. |
Proposed Changes
The Whole Workflow of Job Recovery
...
- Initialize all ExecutionJobVertex whose parallelism has been decided. We can obtain the initialization information from the replayed events (ExecutionJobVertexInitializedEvent).
- According to the information in JobMasterPartitionTracker, the execution vertices whose produced partitions are all tracked will be marked as finished.
- For execution vertices that are not marked as finished, as mentioned above, if its corresponding job vertex has operator coordinators, we need to call subtaskReset for them.
- Find all sink/leaf execution vertices in ExecutionGraph. For each sink/leaf execution vertex in the non-finish state, recursively find all its upstream vertices that need to be restarted (which are in unfinished state), and then start scheduling based on this.
interface ShuffleMaster<T extends ShuffleDescriptor> extends AutoCloseable { //… other methods /** * Get all partitions and their metrics, the metrics mainly includes the meta information of partition(partition bytes, etc). * @param jobId ID of the target job * @return All partitions belongs to the target job and their metrics */ Collection<PartitionWithMetrics> getAllPartitionWithMetrics(JobID jobId); interface PartitionWithMetrics { ShuffleMetrics getPartitionMetrics(); ShuffleDescriptor getPartition(); } interface ShuffleMetrics { ResultPartitionBytes getPartitionBytes(); } } |
Compatibility, Deprecation, and Migration Plan
...