You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "reswqa (via GitHub)" <gi...@apache.org> on 2023/02/08 07:31:04 UTC
[GitHub] [flink] reswqa opened a new pull request, #21890: [FLINK-30860]Add document for hybrid shuffle with adaptive batch scheduler
reswqa opened a new pull request, #21890:
URL: https://github.com/apache/flink/pull/21890
## What is the purpose of the change
*Add document for hybrid shuffle with adaptive batch scheduler*
## Brief change log
- *Add document for hybrid shuffle with adaptive batch scheduler*
- *Add document for hybrid shuffle index spilling*
## Verifying this change
This change is a trivial rework / code cleanup without any test coverage.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #21890: [FLINK-30860][doc] Add document for hybrid shuffle with adaptive batch scheduler
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #21890:
URL: https://github.com/apache/flink/pull/21890#discussion_r1102386579
##########
docs/content.zh/docs/ops/batch/batch_shuffle.md:
##########
@@ -114,12 +114,34 @@ Hybrid shuffle provides two spilling strategies:
To use hybrid shuffle mode, you need to configure the [execution.batch-shuffle-mode]({{< ref "docs/deployment/config" >}}#execution-batch-shuffle-mode) to `ALL_EXCHANGES_HYBRID_FULL` (full spilling strategy) or `ALL_EXCHANGES_HYBRID_SELECTIVE` (selective spilling strategy).
+#### Supports AdaptiveBatchScheduler and SpeculativeExecution
+
+Hybrid shuffle currently supports `AdaptiveBatchScheduler` by default. If you want to use `DefaultScheduler`, please configure the [jobmanager.scheduler]({{< ref "docs/deployment/config" >}}#jobmanager-scheduler) to `DefaultScheduler`. See [elastic_scaling]({{< ref "docs/deployment/elastic_scaling" >}}#adaptive-batch-scheduler) for details.
+
+If you want to enable `SpeculativeExecution` in the same time, see [speculative_execution]({{< ref "docs/deployment/speculative_execution" >}}) for details.
+
+Hybrid shuffle divides the partition data consumption constraints between producer and consumer into the following three cases:
+
+- **ALL_PRODUCERS_FINISHED** : hybrid partition data can be consumed only when all producers are finished.
+- **ONLY_FINISHED_PRODUCERS** : hybrid partition data can be consumed when its producer is finished.
+- **UNFINISHED_PRODUCERS** : hybrid partition data can be consumed even if its producer is un-finished.
+
+If `SpeculativeExecution` is enabled, the default constraint is `ONLY_FINISHED_PRODUCERS` to bring some performance optimization compared with blocking shuffle. Otherwise, the default constraint is `UNFINISHED_PRODUCERS` to perform pipelined-like shuffle. These could be configured via [jobmanager.partition.hybrid.partition-data-consume-constraint]({{< ref "docs/deployment/config" >}}#jobmanager-partition-hybrid-partition-data-consume-constraint).
Review Comment:
Added some descriptions of this part.
##########
docs/content.zh/docs/ops/batch/batch_shuffle.md:
##########
@@ -114,12 +114,34 @@ Hybrid shuffle provides two spilling strategies:
To use hybrid shuffle mode, you need to configure the [execution.batch-shuffle-mode]({{< ref "docs/deployment/config" >}}#execution-batch-shuffle-mode) to `ALL_EXCHANGES_HYBRID_FULL` (full spilling strategy) or `ALL_EXCHANGES_HYBRID_SELECTIVE` (selective spilling strategy).
+#### Supports AdaptiveBatchScheduler and SpeculativeExecution
+
+Hybrid shuffle currently supports `AdaptiveBatchScheduler` by default. If you want to use `DefaultScheduler`, please configure the [jobmanager.scheduler]({{< ref "docs/deployment/config" >}}#jobmanager-scheduler) to `DefaultScheduler`. See [elastic_scaling]({{< ref "docs/deployment/elastic_scaling" >}}#adaptive-batch-scheduler) for details.
+
+If you want to enable `SpeculativeExecution` in the same time, see [speculative_execution]({{< ref "docs/deployment/speculative_execution" >}}) for details.
Review Comment:
Removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a diff in pull request #21890: [FLINK-30860][doc] Add document for hybrid shuffle with adaptive batch scheduler
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #21890:
URL: https://github.com/apache/flink/pull/21890#discussion_r1105226779
##########
docs/content/docs/ops/batch/batch_shuffle.md:
##########
@@ -112,14 +112,27 @@ Hybrid shuffle provides two spilling strategies:
### Usage
-To use hybrid shuffle mode, you need to configure the [execution.batch-shuffle-mode]({{< ref "docs/deployment/config" >}}#execution.batch-shuffle-mode) to `ALL_EXCHANGES_HYBRID_FULL` (full spilling strategy) or `ALL_EXCHANGES_HYBRID_SELECTIVE` (selective spilling strategy).
+To use hybrid shuffle mode, you need to configure the [execution.batch-shuffle-mode]({{< ref "docs/deployment/config" >}}#execution-batch-shuffle-mode) to `ALL_EXCHANGES_HYBRID_FULL` (full spilling strategy) or `ALL_EXCHANGES_HYBRID_SELECTIVE` (selective spilling strategy).
+
+#### Data Consumption Constraints
+
+Hybrid shuffle divides the partition data consumption constraints between producer and consumer into the following three cases:
+
+- **ALL_PRODUCERS_FINISHED** : hybrid partition data can be consumed only when all producers are finished.
+- **ONLY_FINISHED_PRODUCERS** : hybrid partition data can be consumed when its producer is finished.
+- **UNFINISHED_PRODUCERS** : hybrid partition data can be consumed even if its producer is un-finished.
+
+These could be configured via [jobmanager.partition.hybrid.partition-data-consume-constraint]({{< ref "docs/deployment/config" >}}#jobmanager-partition-hybrid-partition-data-consume-constraint).
+
+- **For `AdaptiveBatchScheduler`** : The default constraint is `UNFINISHED_PRODUCERS` to perform pipelined-like shuffle. If the value is set to `ALL_PRODUCERS_FINISHED` or `ONLY_FINISHED_PRODUCERS`, performance may be degraded.
+- **If `SpeculativeExecution` is enabled** : The default constraint is `ONLY_FINISHED_PRODUCERS` to bring some performance optimization compared with blocking shuffle. Since producers and consumers have the opportunity to run at the same time, more speculative execution tasks may be created, and the cost of failover will also increase. If you want to fall back to the same behavior as blocking shuffle, you can configure this value to `ALL_PRODUCERS_FINISHED`. It is also important to note that `UNFINISHED_PRODUCERS` is not supported in this mode.
### Limitations
Hybrid shuffle mode is still experimental and has some known limitations, which the Flink community is still working on eliminating.
- **No support for Slot Sharing.** In hybrid shuffle mode, Flink currently forces each task to be executed in a dedicated slot exclusively. If slot sharing is explicitly specified, an error will occur.
-- **No support for Adaptive Batch Scheduler and Speculative Execution.** If adaptive batch scheduler is used in hybrid shuffle mode, an error will occur.
+- **No optimization for dynamic graph.** If auto-parallelism(dynamic graph) is enabled for `AdaptiveBatchScheduler`, hybrid shuffle will always schedule tasks only when all producer are finished like blocking shuffle, this means that the constraint will fall back to `ALL_PRODUCERS_FINISHED` in this case.
Review Comment:
```suggestion
- **No pipelined execution for dynamic graph.** If auto-parallelism (dynamic graph) is enabled, Adaptive Batch Scheduler will wait until upstream tasks finish to decide parallelism of downstream tasks, which means hybrid shuffle effectively fallback to blocking shuffle (`ALL_PRODUCERS_FINISHED` constraint).
```
##########
docs/content/docs/ops/batch/batch_shuffle.md:
##########
@@ -112,14 +112,27 @@ Hybrid shuffle provides two spilling strategies:
### Usage
-To use hybrid shuffle mode, you need to configure the [execution.batch-shuffle-mode]({{< ref "docs/deployment/config" >}}#execution.batch-shuffle-mode) to `ALL_EXCHANGES_HYBRID_FULL` (full spilling strategy) or `ALL_EXCHANGES_HYBRID_SELECTIVE` (selective spilling strategy).
+To use hybrid shuffle mode, you need to configure the [execution.batch-shuffle-mode]({{< ref "docs/deployment/config" >}}#execution-batch-shuffle-mode) to `ALL_EXCHANGES_HYBRID_FULL` (full spilling strategy) or `ALL_EXCHANGES_HYBRID_SELECTIVE` (selective spilling strategy).
+
+#### Data Consumption Constraints
+
+Hybrid shuffle divides the partition data consumption constraints between producer and consumer into the following three cases:
+
+- **ALL_PRODUCERS_FINISHED** : hybrid partition data can be consumed only when all producers are finished.
+- **ONLY_FINISHED_PRODUCERS** : hybrid partition data can be consumed when its producer is finished.
+- **UNFINISHED_PRODUCERS** : hybrid partition data can be consumed even if its producer is un-finished.
Review Comment:
```suggestion
- **ONLY_FINISHED_PRODUCERS** : hybrid partition can only consume data from finished producers.
- **UNFINISHED_PRODUCERS** : hybrid partition can consume data from unfinished producers.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a diff in pull request #21890: [FLINK-30860][doc] Add document for hybrid shuffle with adaptive batch scheduler
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #21890:
URL: https://github.com/apache/flink/pull/21890#discussion_r1101054167
##########
docs/content.zh/docs/ops/batch/batch_shuffle.md:
##########
@@ -114,12 +114,34 @@ Hybrid shuffle provides two spilling strategies:
To use hybrid shuffle mode, you need to configure the [execution.batch-shuffle-mode]({{< ref "docs/deployment/config" >}}#execution-batch-shuffle-mode) to `ALL_EXCHANGES_HYBRID_FULL` (full spilling strategy) or `ALL_EXCHANGES_HYBRID_SELECTIVE` (selective spilling strategy).
+#### Supports AdaptiveBatchScheduler and SpeculativeExecution
+
+Hybrid shuffle currently supports `AdaptiveBatchScheduler` by default. If you want to use `DefaultScheduler`, please configure the [jobmanager.scheduler]({{< ref "docs/deployment/config" >}}#jobmanager-scheduler) to `DefaultScheduler`. See [elastic_scaling]({{< ref "docs/deployment/elastic_scaling" >}}#adaptive-batch-scheduler) for details.
+
+If you want to enable `SpeculativeExecution` in the same time, see [speculative_execution]({{< ref "docs/deployment/speculative_execution" >}}) for details.
Review Comment:
This is irrelevant to hybrid shuffle.
##########
docs/content.zh/docs/ops/batch/batch_shuffle.md:
##########
@@ -114,12 +114,34 @@ Hybrid shuffle provides two spilling strategies:
To use hybrid shuffle mode, you need to configure the [execution.batch-shuffle-mode]({{< ref "docs/deployment/config" >}}#execution-batch-shuffle-mode) to `ALL_EXCHANGES_HYBRID_FULL` (full spilling strategy) or `ALL_EXCHANGES_HYBRID_SELECTIVE` (selective spilling strategy).
+#### Supports AdaptiveBatchScheduler and SpeculativeExecution
+
+Hybrid shuffle currently supports `AdaptiveBatchScheduler` by default. If you want to use `DefaultScheduler`, please configure the [jobmanager.scheduler]({{< ref "docs/deployment/config" >}}#jobmanager-scheduler) to `DefaultScheduler`. See [elastic_scaling]({{< ref "docs/deployment/elastic_scaling" >}}#adaptive-batch-scheduler) for details.
+
+If you want to enable `SpeculativeExecution` in the same time, see [speculative_execution]({{< ref "docs/deployment/speculative_execution" >}}) for details.
+
+Hybrid shuffle divides the partition data consumption constraints between producer and consumer into the following three cases:
+
+- **ALL_PRODUCERS_FINISHED** : hybrid partition data can be consumed only when all producers are finished.
+- **ONLY_FINISHED_PRODUCERS** : hybrid partition data can be consumed when its producer is finished.
+- **UNFINISHED_PRODUCERS** : hybrid partition data can be consumed even if its producer is un-finished.
+
+If `SpeculativeExecution` is enabled, the default constraint is `ONLY_FINISHED_PRODUCERS` to bring some performance optimization compared with blocking shuffle. Otherwise, the default constraint is `UNFINISHED_PRODUCERS` to perform pipelined-like shuffle. These could be configured via [jobmanager.partition.hybrid.partition-data-consume-constraint]({{< ref "docs/deployment/config" >}}#jobmanager-partition-hybrid-partition-data-consume-constraint).
Review Comment:
What is the potential impacts when changing this option?
##########
docs/content.zh/docs/ops/batch/batch_shuffle.md:
##########
@@ -114,12 +114,34 @@ Hybrid shuffle provides two spilling strategies:
To use hybrid shuffle mode, you need to configure the [execution.batch-shuffle-mode]({{< ref "docs/deployment/config" >}}#execution-batch-shuffle-mode) to `ALL_EXCHANGES_HYBRID_FULL` (full spilling strategy) or `ALL_EXCHANGES_HYBRID_SELECTIVE` (selective spilling strategy).
+#### Supports AdaptiveBatchScheduler and SpeculativeExecution
+
+Hybrid shuffle currently supports `AdaptiveBatchScheduler` by default. If you want to use `DefaultScheduler`, please configure the [jobmanager.scheduler]({{< ref "docs/deployment/config" >}}#jobmanager-scheduler) to `DefaultScheduler`. See [elastic_scaling]({{< ref "docs/deployment/elastic_scaling" >}}#adaptive-batch-scheduler) for details.
+
+If you want to enable `SpeculativeExecution` in the same time, see [speculative_execution]({{< ref "docs/deployment/speculative_execution" >}}) for details.
+
+Hybrid shuffle divides the partition data consumption constraints between producer and consumer into the following three cases:
+
+- **ALL_PRODUCERS_FINISHED** : hybrid partition data can be consumed only when all producers are finished.
+- **ONLY_FINISHED_PRODUCERS** : hybrid partition data can be consumed when its producer is finished.
+- **UNFINISHED_PRODUCERS** : hybrid partition data can be consumed even if its producer is un-finished.
+
+If `SpeculativeExecution` is enabled, the default constraint is `ONLY_FINISHED_PRODUCERS` to bring some performance optimization compared with blocking shuffle. Otherwise, the default constraint is `UNFINISHED_PRODUCERS` to perform pipelined-like shuffle. These could be configured via [jobmanager.partition.hybrid.partition-data-consume-constraint]({{< ref "docs/deployment/config" >}}#jobmanager-partition-hybrid-partition-data-consume-constraint).
+
+#### Index Spilling
+
+Hybrid shuffle indexes the shuffle data in memory and disk. Generally speaking, all index can be cached in memory to speed up index retrieval. However, for large batch jobs, this part of memory may bring OOM risks.
+Therefore, hybrid shuffle supports spilling index data to disk. The following configuration options can control this behavior:
+
+- **[taskmanager.network.hybrid-shuffle.num-retained-in-memory-regions-max]({{< ref "docs/deployment/config" >}}#taskmanager-network-hybrid-shuffle-num-retained-in-memory-regions-max)** : Controls the max number of hybrid retained regions in memory. Increasing this value will allow more index entries to be cached in memory.
+- **[taskmanager.network.hybrid-shuffle.spill-index-segment-size]({{< ref "docs/deployment/config" >}}#taskmanager-network-hybrid-shuffle-spill-index-segment-size)** : Controls the segment size(in bytes) of hybrid spilled file data index.
Review Comment:
I wonder if we want to advertise these options. They are likely removed in future releases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #21890: [FLINK-30860][doc] Add document for hybrid shuffle with adaptive batch scheduler
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #21890:
URL: https://github.com/apache/flink/pull/21890#discussion_r1102385798
##########
docs/content.zh/docs/ops/batch/batch_shuffle.md:
##########
@@ -114,12 +114,34 @@ Hybrid shuffle provides two spilling strategies:
To use hybrid shuffle mode, you need to configure the [execution.batch-shuffle-mode]({{< ref "docs/deployment/config" >}}#execution-batch-shuffle-mode) to `ALL_EXCHANGES_HYBRID_FULL` (full spilling strategy) or `ALL_EXCHANGES_HYBRID_SELECTIVE` (selective spilling strategy).
+#### Supports AdaptiveBatchScheduler and SpeculativeExecution
+
+Hybrid shuffle currently supports `AdaptiveBatchScheduler` by default. If you want to use `DefaultScheduler`, please configure the [jobmanager.scheduler]({{< ref "docs/deployment/config" >}}#jobmanager-scheduler) to `DefaultScheduler`. See [elastic_scaling]({{< ref "docs/deployment/elastic_scaling" >}}#adaptive-batch-scheduler) for details.
+
+If you want to enable `SpeculativeExecution` in the same time, see [speculative_execution]({{< ref "docs/deployment/speculative_execution" >}}) for details.
+
+Hybrid shuffle divides the partition data consumption constraints between producer and consumer into the following three cases:
+
+- **ALL_PRODUCERS_FINISHED** : hybrid partition data can be consumed only when all producers are finished.
+- **ONLY_FINISHED_PRODUCERS** : hybrid partition data can be consumed when its producer is finished.
+- **UNFINISHED_PRODUCERS** : hybrid partition data can be consumed even if its producer is un-finished.
+
+If `SpeculativeExecution` is enabled, the default constraint is `ONLY_FINISHED_PRODUCERS` to bring some performance optimization compared with blocking shuffle. Otherwise, the default constraint is `UNFINISHED_PRODUCERS` to perform pipelined-like shuffle. These could be configured via [jobmanager.partition.hybrid.partition-data-consume-constraint]({{< ref "docs/deployment/config" >}}#jobmanager-partition-hybrid-partition-data-consume-constraint).
+
+#### Index Spilling
+
+Hybrid shuffle indexes the shuffle data in memory and disk. Generally speaking, all index can be cached in memory to speed up index retrieval. However, for large batch jobs, this part of memory may bring OOM risks.
+Therefore, hybrid shuffle supports spilling index data to disk. The following configuration options can control this behavior:
+
+- **[taskmanager.network.hybrid-shuffle.num-retained-in-memory-regions-max]({{< ref "docs/deployment/config" >}}#taskmanager-network-hybrid-shuffle-num-retained-in-memory-regions-max)** : Controls the max number of hybrid retained regions in memory. Increasing this value will allow more index entries to be cached in memory.
+- **[taskmanager.network.hybrid-shuffle.spill-index-segment-size]({{< ref "docs/deployment/config" >}}#taskmanager-network-hybrid-shuffle-spill-index-segment-size)** : Controls the segment size(in bytes) of hybrid spilled file data index.
Review Comment:
Agree with you, removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #21890: [FLINK-30860]Add document for hybrid shuffle with adaptive batch scheduler
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #21890:
URL: https://github.com/apache/flink/pull/21890#issuecomment-1422164243
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8aeaf0001bf4c156427acc1202a060507871532e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8aeaf0001bf4c156427acc1202a060507871532e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8aeaf0001bf4c156427acc1202a060507871532e UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on pull request #21890: [FLINK-30860][doc] Add document for hybrid shuffle with adaptive batch scheduler
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on PR #21890:
URL: https://github.com/apache/flink/pull/21890#issuecomment-1425419929
Thanks @xintongsong for the review, I have addressed all comments, PTAL~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa closed pull request #21890: [FLINK-30860][doc] Add document for hybrid shuffle with adaptive batch scheduler
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa closed pull request #21890: [FLINK-30860][doc] Add document for hybrid shuffle with adaptive batch scheduler
URL: https://github.com/apache/flink/pull/21890
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org