This is an automated email from the ASF dual-hosted git repository. sewen pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 11ba5b3b26e146d80e1319d77d1ee5ef2dd261d1 Author: kevin.cyj <kevin....@alibaba-inc.com> AuthorDate: Mon Oct 12 20:45:49 2020 +0800 [FLINK-19602][network] Introduce new config options to enable sort-merge based blocking shuffle Two new config options are added to control the behavior of the sort-merge based blocking shuffle: 1. taskmanager.network.sort-shuffle.min-buffers: Minimum number of network buffers required per sort-merge blocking result partition. 2. taskmanager.network.sort-shuffle.min-parallelism: Parallelism threshold to switch between sort-merge blocking shuffle and the default hash-based blocking shuffle. With these new config options, the default behavior of blocking shuffle stays unchanged. --- .../generated/all_taskmanager_network_section.html | 12 ++ .../netty_shuffle_environment_configuration.html | 12 ++ docs/ops/memory/mem_tuning.md | 20 ++++ docs/ops/memory/mem_tuning.zh.md | 17 +++ .../NettyShuffleEnvironmentOptions.java | 30 +++++ .../io/network/NettyShuffleServiceFactory.java | 4 +- .../network/partition/ResultPartitionFactory.java | 68 ++++++++---- .../NettyShuffleEnvironmentConfiguration.java | 33 +++++- .../io/network/NettyShuffleEnvironmentBuilder.java | 18 ++- .../network/partition/ResultPartitionBuilder.java | 22 +++- .../partition/ResultPartitionFactoryTest.java | 29 ++++- .../flink/test/runtime/BlockingShuffleITCase.java | 123 +++++++++++++++++++++ .../flink/test/runtime/JobGraphRunningUtil.java | 62 +++++++++++ .../test/runtime/ShuffleCompressionITCase.java | 33 +----- 14 files changed, 423 insertions(+), 60 deletions(-) diff --git a/docs/_includes/generated/all_taskmanager_network_section.html b/docs/_includes/generated/all_taskmanager_network_section.html index 6c2a11b..5e5336d 100644 --- a/docs/_includes/generated/all_taskmanager_network_section.html +++ b/docs/_includes/generated/all_taskmanager_network_section.html @@ -104,5 +104,17 @@ <td>Integer</td> <td>The number of retry attempts for network communication. Currently it's only used for establishing input/output channel connections</td> </tr> + <tr> + <td><h5>taskmanager.network.sort-shuffle.min-buffers</h5></td> + <td style="word-wrap: break-word;">64</td> + <td>Integer</td> + <td>Minimum number of network buffers required per sort-merge blocking result partition. For large scale batch jobs, it is suggested to increase this config value to improve compression ratio and reduce small network packets. Note: to increase this config value, you may also need to increase the size of total network memory to avoid "insufficient number of network buffers" error.</td> + </tr> + <tr> + <td><h5>taskmanager.network.sort-shuffle.min-parallelism</h5></td> + <td style="word-wrap: break-word;">2147483647</td> + <td>Integer</td> + <td>Parallelism threshold to switch between sort-merge blocking shuffle and the default hash-based blocking shuffle, which means for small parallelism, hash-based blocking shuffle will be used and for large parallelism, sort-merge blocking shuffle will be used. Note: sort-merge blocking shuffle uses unmanaged direct memory for shuffle data writing and reading so just increase the size of direct memory if direct memory OOM error occurs.</td> + </tr> </tbody> </table> diff --git a/docs/_includes/generated/netty_shuffle_environment_configuration.html b/docs/_includes/generated/netty_shuffle_environment_configuration.html index 018296a..9d54dca 100644 --- a/docs/_includes/generated/netty_shuffle_environment_configuration.html +++ b/docs/_includes/generated/netty_shuffle_environment_configuration.html @@ -122,5 +122,17 @@ <td>Integer</td> <td>The number of retry attempts for network communication. Currently it's only used for establishing input/output channel connections</td> </tr> + <tr> + <td><h5>taskmanager.network.sort-shuffle.min-buffers</h5></td> + <td style="word-wrap: break-word;">64</td> + <td>Integer</td> + <td>Minimum number of network buffers required per sort-merge blocking result partition. For large scale batch jobs, it is suggested to increase this config value to improve compression ratio and reduce small network packets. Note: to increase this config value, you may also need to increase the size of total network memory to avoid "insufficient number of network buffers" error.</td> + </tr> + <tr> + <td><h5>taskmanager.network.sort-shuffle.min-parallelism</h5></td> + <td style="word-wrap: break-word;">2147483647</td> + <td>Integer</td> + <td>Parallelism threshold to switch between sort-merge blocking shuffle and the default hash-based blocking shuffle, which means for small parallelism, hash-based blocking shuffle will be used and for large parallelism, sort-merge blocking shuffle will be used. Note: sort-merge blocking shuffle uses unmanaged direct memory for shuffle data writing and reading so just increase the size of direct memory if direct memory OOM error occurs.</td> + </tr> </tbody> </table> diff --git a/docs/ops/memory/mem_tuning.md b/docs/ops/memory/mem_tuning.md index bbb210c..2cbbdbc 100644 --- a/docs/ops/memory/mem_tuning.md +++ b/docs/ops/memory/mem_tuning.md @@ -88,3 +88,23 @@ on the performance of your applications. Flink will attempt to allocate and use as configured for batch jobs but not go beyond its limits. This prevents `OutOfMemoryError`'s because Flink knows precisely how much memory it has to leverage. If the [managed memory](../memory/mem_setup_tm.html#managed-memory) is not sufficient, Flink will gracefully spill to disk. + +## Configure memory for sort-merge blocking shuffle + +The number of required network buffers per sort-merge blocking result partition is controlled by +[taskmanager.network.sort-shuffle.min-buffers](../config.html#taskmanager-network-sort-shuffle-min-buffers) +and the default value is 64 which is quite small. Though it can work for arbitrary parallelism, the +performance may not be the best. For large scale jobs, it is suggested to increase this config value +to improve compression ratio and reduce small network packets which is good for performance. To increase +this value, you may also need to increase the size of total network memory by adjusting the config +values of [taskmanager.memory.network.fraction](../config.html#taskmanager-memory-network-fraction), +[taskmanager.memory.network.min](../config.html#taskmanager-memory-network-min) and [taskmanager. +memory.network.max](../config.html#taskmanager-memory-network-max) to avoid `insufficient number of +network buffers` error. + +Except for network memory, the sort-merge blocking shuffle implementation also uses some unmanaged +direct memory for shuffle data writing and reading. So to use sort-merge blocking shuffle, you may +need to reserve some direct memory for it by increasing the config value of [taskmanager.memory.task +.off-heap.size](../config.html#taskmanager-memory-task-off-heap-size). If direct memory OOM error +occurs after you enable the sort-merge blocking shuffle, you can just give more direct memory until +the OOM error disappears. diff --git a/docs/ops/memory/mem_tuning.zh.md b/docs/ops/memory/mem_tuning.zh.md index e4f8c99..1533012 100644 --- a/docs/ops/memory/mem_tuning.zh.md +++ b/docs/ops/memory/mem_tuning.zh.md @@ -85,3 +85,20 @@ Flink 批处理算子使用[托管内存](../memory/mem_setup_tm.html#managed-me 因此 Flink 会在不超过其配置限额的前提下,尽可能分配更多的[托管内存](../memory/mem_setup_tm.html#managed-memory)。 Flink 明确知道可以使用的内存大小,因此可以有效避免 `OutOfMemoryError` 的发生。 当[托管内存](../memory/mem_setup_tm.html#managed-memory)不足时,Flink 会优雅地将数据落盘。 + +## SortMerge数据Shuffle内存配置 + +对于SortMerge数据Shuffle,每个ResultPartition需要的网络缓冲区(Buffer)数目是由[taskmanager.network.sort- +shuffle.min-buffers](../config.html#taskmanager-network-sort-shuffle-min-buffers)这个配置决定的。它的 +默认值是64,是比较小的。虽然64个网络Buffer已经可以支持任意规模的并发,但性能可能不是最好的。对于大并发的作业,通 +过增大这个配置值,可以提高落盘数据的压缩率并且减少网络小包的数量,从而有利于提高Shuffle性能。为了增大这个配置值, +你可能需要通过调整[taskmanager.memory.network.fraction](../config.html#taskmanager-memory-network-fraction), +[taskmanager.memory.network.min](../config.html#taskmanager-memory-network-min)和[taskmanager.memory +.network.max](../config.html#taskmanager-memory-network-max)这三个参数来增大总的网络内存大小从而避免出现 +`insufficient number of network buffers`错误。 + +除了网络内存,SortMerge数据Shuffle还需要使用一些JVM Direct Memory来进行Shuffle数据的写出与读取。所以,为了使 +用SortMerge数据Shuffle你可能还需要通过增大这个配置值[taskmanager.memory.task.off-heap.size +](../config.html#taskmanager-memory-task-off-heap-size)来为其来预留一些JVM Direct Memory。如果在你开启 +SortMerge数据Shuffle之后出现了Direct Memory OOM的错误,你只需要继续加大上面的配置值来预留更多的Direct Memory +直到不再发生Direct Memory OOM的错误为止。 diff --git a/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java index 584bc6e..f8c71c6 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java @@ -174,6 +174,36 @@ public class NettyShuffleEnvironmentOptions { " increased in case of higher round trip times between nodes and/or larger number of machines in the cluster."); /** + * Minimum number of network buffers required per sort-merge blocking result partition. + */ + @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK) + public static final ConfigOption<Integer> NETWORK_SORT_SHUFFLE_MIN_BUFFERS = + key("taskmanager.network.sort-shuffle.min-buffers") + .intType() + .defaultValue(64) + .withDescription("Minimum number of network buffers required per sort-merge blocking " + + "result partition. For large scale batch jobs, it is suggested to increase this" + + " config value to improve compression ratio and reduce small network packets. " + + "Note: to increase this config value, you may also need to increase the size of " + + "total network memory to avoid \"insufficient number of network buffers\" error."); + + /** + * Parallelism threshold to switch between sort-merge based blocking shuffle and the default + * hash-based blocking shuffle. + */ + @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK) + public static final ConfigOption<Integer> NETWORK_SORT_SHUFFLE_MIN_PARALLELISM = + key("taskmanager.network.sort-shuffle.min-parallelism") + .intType() + .defaultValue(Integer.MAX_VALUE) + .withDescription("Parallelism threshold to switch between sort-merge blocking shuffle " + + "and the default hash-based blocking shuffle, which means for small parallelism," + + " hash-based blocking shuffle will be used and for large parallelism, sort-merge" + + " blocking shuffle will be used. Note: sort-merge blocking shuffle uses unmanaged" + + " direct memory for shuffle data writing and reading so just increase the size of" + + " direct memory if direct memory OOM error occurs."); + + /** * Number of max buffers can be used for each output subparition. */ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NettyShuffleServiceFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NettyShuffleServiceFactory.java index d5863d7..ecd8027 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NettyShuffleServiceFactory.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NettyShuffleServiceFactory.java @@ -126,7 +126,9 @@ public class NettyShuffleServiceFactory implements ShuffleServiceFactory<NettySh config.networkBufferSize(), config.isBlockingShuffleCompressionEnabled(), config.getCompressionCodec(), - config.getMaxBuffersPerChannel()); + config.getMaxBuffersPerChannel(), + config.sortShuffleMinBuffers(), + config.sortShuffleMinParallelism()); SingleInputGateFactory singleInputGateFactory = new SingleInputGateFactory( taskExecutorResourceId, diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java index c6a3a4e..c83eaff 100755 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java @@ -64,6 +64,10 @@ public class ResultPartitionFactory { private final int maxBuffersPerChannel; + private final int sortShuffleMinBuffers; + + private final int sortShuffleMinParallelism; + public ResultPartitionFactory( ResultPartitionManager partitionManager, FileChannelManager channelManager, @@ -74,7 +78,9 @@ public class ResultPartitionFactory { int networkBufferSize, boolean blockingShuffleCompressionEnabled, String compressionCodec, - int maxBuffersPerChannel) { + int maxBuffersPerChannel, + int sortShuffleMinBuffers, + int sortShuffleMinParallelism) { this.partitionManager = partitionManager; this.channelManager = channelManager; @@ -86,6 +92,8 @@ public class ResultPartitionFactory { this.blockingShuffleCompressionEnabled = blockingShuffleCompressionEnabled; this.compressionCodec = compressionCodec; this.maxBuffersPerChannel = maxBuffersPerChannel; + this.sortShuffleMinBuffers = sortShuffleMinBuffers; + this.sortShuffleMinParallelism = sortShuffleMinParallelism; } public ResultPartition create( @@ -146,25 +154,40 @@ public class ResultPartitionFactory { partition = pipelinedPartition; } else if (type == ResultPartitionType.BLOCKING || type == ResultPartitionType.BLOCKING_PERSISTENT) { - final BoundedBlockingResultPartition blockingPartition = new BoundedBlockingResultPartition( - taskNameWithSubtaskAndId, - partitionIndex, - id, - type, - subpartitions, - maxParallelism, - partitionManager, - bufferCompressor, - bufferPoolFactory); - - initializeBoundedBlockingPartitions( - subpartitions, - blockingPartition, - blockingSubpartitionType, - networkBufferSize, - channelManager); - - partition = blockingPartition; + if (numberOfSubpartitions >= sortShuffleMinParallelism) { + partition = new SortMergeResultPartition( + taskNameWithSubtaskAndId, + partitionIndex, + id, + type, + subpartitions.length, + maxParallelism, + networkBufferSize, + partitionManager, + channelManager.createChannel().getPath(), + bufferCompressor, + bufferPoolFactory); + } else { + final BoundedBlockingResultPartition blockingPartition = new BoundedBlockingResultPartition( + taskNameWithSubtaskAndId, + partitionIndex, + id, + type, + subpartitions, + maxParallelism, + partitionManager, + bufferCompressor, + bufferPoolFactory); + + initializeBoundedBlockingPartitions( + subpartitions, + blockingPartition, + blockingSubpartitionType, + networkBufferSize, + channelManager); + + partition = blockingPartition; + } } else { throw new IllegalArgumentException("Unrecognized ResultPartitionType: " + type); @@ -224,10 +247,13 @@ public class ResultPartitionFactory { return () -> { int maxNumberOfMemorySegments = type.isBounded() ? numberOfSubpartitions * networkBuffersPerChannel + floatingNetworkBuffersPerGate : Integer.MAX_VALUE; + int numRequiredBuffers = !type.isPipelined() && numberOfSubpartitions >= sortShuffleMinParallelism ? + sortShuffleMinBuffers : numberOfSubpartitions + 1; + // If the partition type is back pressure-free, we register with the buffer pool for // callbacks to release memory. return bufferPoolFactory.createBufferPool( - numberOfSubpartitions + 1, + numRequiredBuffers, maxNumberOfMemorySegments, numberOfSubpartitions, maxBuffersPerChannel); diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java index 09505dc..8f32660 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java @@ -58,6 +58,10 @@ public class NettyShuffleEnvironmentConfiguration { /** Number of extra network buffers to use for each outgoing/incoming gate (result partition/input gate). */ private final int floatingNetworkBuffersPerGate; + private final int sortShuffleMinBuffers; + + private final int sortShuffleMinParallelism; + private final Duration requestSegmentsTimeout; private final boolean isNetworkDetailedMetrics; @@ -88,7 +92,9 @@ public class NettyShuffleEnvironmentConfiguration { BoundedBlockingSubpartitionType blockingSubpartitionType, boolean blockingShuffleCompressionEnabled, String compressionCodec, - int maxBuffersPerChannel) { + int maxBuffersPerChannel, + int sortShuffleMinBuffers, + int sortShuffleMinParallelism) { this.numNetworkBuffers = numNetworkBuffers; this.networkBufferSize = networkBufferSize; @@ -104,6 +110,8 @@ public class NettyShuffleEnvironmentConfiguration { this.blockingShuffleCompressionEnabled = blockingShuffleCompressionEnabled; this.compressionCodec = Preconditions.checkNotNull(compressionCodec); this.maxBuffersPerChannel = maxBuffersPerChannel; + this.sortShuffleMinBuffers = sortShuffleMinBuffers; + this.sortShuffleMinParallelism = sortShuffleMinParallelism; } // ------------------------------------------------------------------------ @@ -132,6 +140,14 @@ public class NettyShuffleEnvironmentConfiguration { return floatingNetworkBuffersPerGate; } + public int sortShuffleMinBuffers() { + return sortShuffleMinBuffers; + } + + public int sortShuffleMinParallelism() { + return sortShuffleMinParallelism; + } + public Duration getRequestSegmentsTimeout() { return requestSegmentsTimeout; } @@ -201,6 +217,11 @@ public class NettyShuffleEnvironmentConfiguration { int maxBuffersPerChannel = configuration.getInteger(NettyShuffleEnvironmentOptions.NETWORK_MAX_BUFFERS_PER_CHANNEL); + int sortShuffleMinBuffers = configuration.getInteger( + NettyShuffleEnvironmentOptions.NETWORK_SORT_SHUFFLE_MIN_BUFFERS); + int sortShuffleMinParallelism = configuration.getInteger( + NettyShuffleEnvironmentOptions.NETWORK_SORT_SHUFFLE_MIN_PARALLELISM); + boolean isNetworkDetailedMetrics = configuration.getBoolean(NettyShuffleEnvironmentOptions.NETWORK_DETAILED_METRICS); String[] tempDirs = ConfigurationUtils.parseTempDirectories(configuration); @@ -228,7 +249,9 @@ public class NettyShuffleEnvironmentConfiguration { blockingSubpartitionType, blockingShuffleCompressionEnabled, compressionCodec, - maxBuffersPerChannel); + maxBuffersPerChannel, + sortShuffleMinBuffers, + sortShuffleMinParallelism); } /** @@ -350,6 +373,8 @@ public class NettyShuffleEnvironmentConfiguration { result = 31 * result + (blockingShuffleCompressionEnabled ? 1 : 0); result = 31 * result + Objects.hashCode(compressionCodec); result = 31 * result + maxBuffersPerChannel; + result = 31 * result + sortShuffleMinBuffers; + result = 31 * result + sortShuffleMinParallelism; return result; } @@ -370,6 +395,8 @@ public class NettyShuffleEnvironmentConfiguration { this.partitionRequestMaxBackoff == that.partitionRequestMaxBackoff && this.networkBuffersPerChannel == that.networkBuffersPerChannel && this.floatingNetworkBuffersPerGate == that.floatingNetworkBuffersPerGate && + this.sortShuffleMinBuffers == that.sortShuffleMinBuffers && + this.sortShuffleMinParallelism == that.sortShuffleMinParallelism && this.requestSegmentsTimeout.equals(that.requestSegmentsTimeout) && (nettyConfig != null ? nettyConfig.equals(that.nettyConfig) : that.nettyConfig == null) && Arrays.equals(this.tempDirs, that.tempDirs) && @@ -394,6 +421,8 @@ public class NettyShuffleEnvironmentConfiguration { ", blockingShuffleCompressionEnabled=" + blockingShuffleCompressionEnabled + ", compressionCodec=" + compressionCodec + ", maxBuffersPerChannel=" + maxBuffersPerChannel + + ", sortShuffleMinBuffers=" + sortShuffleMinBuffers + + ", sortShuffleMinParallelism=" + sortShuffleMinParallelism + '}'; } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java index 1d625fc..08d2481d 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java @@ -54,6 +54,10 @@ public class NettyShuffleEnvironmentBuilder { private int floatingNetworkBuffersPerGate = 8; + private int sortShuffleMinBuffers = 100; + + private int sortShuffleMinParallelism = Integer.MAX_VALUE; + private int maxBuffersPerChannel = Integer.MAX_VALUE; private boolean blockingShuffleCompressionEnabled = false; @@ -110,6 +114,16 @@ public class NettyShuffleEnvironmentBuilder { return this; } + public NettyShuffleEnvironmentBuilder setSortShuffleMinBuffers(int sortShuffleMinBuffers) { + this.sortShuffleMinBuffers = sortShuffleMinBuffers; + return this; + } + + public NettyShuffleEnvironmentBuilder setSortShuffleMinParallelism(int sortShuffleMinParallelism) { + this.sortShuffleMinParallelism = sortShuffleMinParallelism; + return this; + } + public NettyShuffleEnvironmentBuilder setBlockingShuffleCompressionEnabled(boolean blockingShuffleCompressionEnabled) { this.blockingShuffleCompressionEnabled = blockingShuffleCompressionEnabled; return this; @@ -156,7 +170,9 @@ public class NettyShuffleEnvironmentBuilder { BoundedBlockingSubpartitionType.AUTO, blockingShuffleCompressionEnabled, compressionCodec, - maxBuffersPerChannel), + maxBuffersPerChannel, + sortShuffleMinBuffers, + sortShuffleMinParallelism), taskManagerLocation, new TaskEventDispatcher(), resultPartitionManager, diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java index 826df0b..67e7adb 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java @@ -55,6 +55,10 @@ public class ResultPartitionBuilder { private int floatingNetworkBuffersPerGate = 1; + private int sortShuffleMinBuffers = 100; + + private int sortShuffleMinParallelism = Integer.MAX_VALUE; + private int maxBuffersPerChannel = Integer.MAX_VALUE; private int networkBufferSize = 1; @@ -105,7 +109,9 @@ public class ResultPartitionBuilder { return setNetworkBuffersPerChannel(environment.getConfiguration().networkBuffersPerChannel()) .setFloatingNetworkBuffersPerGate(environment.getConfiguration().floatingNetworkBuffersPerGate()) .setNetworkBufferSize(environment.getConfiguration().networkBufferSize()) - .setNetworkBufferPool(environment.getNetworkBufferPool()); + .setNetworkBufferPool(environment.getNetworkBufferPool()) + .setSortShuffleMinBuffers(environment.getConfiguration().sortShuffleMinBuffers()) + .setSortShuffleMinParallelism(environment.getConfiguration().sortShuffleMinParallelism()); } public ResultPartitionBuilder setNetworkBufferPool(NetworkBufferPool networkBufferPool) { @@ -139,6 +145,16 @@ public class ResultPartitionBuilder { return this; } + public ResultPartitionBuilder setSortShuffleMinBuffers(int sortShuffleMinBuffers) { + this.sortShuffleMinBuffers = sortShuffleMinBuffers; + return this; + } + + public ResultPartitionBuilder setSortShuffleMinParallelism(int sortShuffleMinParallelism) { + this.sortShuffleMinParallelism = sortShuffleMinParallelism; + return this; + } + public ResultPartitionBuilder setCompressionCodec(String compressionCodec) { this.compressionCodec = compressionCodec; return this; @@ -161,7 +177,9 @@ public class ResultPartitionBuilder { networkBufferSize, blockingShuffleCompressionEnabled, compressionCodec, - maxBuffersPerChannel); + maxBuffersPerChannel, + sortShuffleMinBuffers, + sortShuffleMinParallelism); SupplierWithException<BufferPool, IOException> factory = bufferPoolFactory.orElseGet(() -> resultPartitionFactory.createBufferPoolFactory(numberOfSubpartitions, partitionType)); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java index 903878b..40297f6 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java @@ -72,7 +72,13 @@ public class ResultPartitionFactoryTest extends TestLogger { } @Test - public void testConsumptionOnReleaseForPipelined() { + public void testSortMergePartitionCreated() { + ResultPartition resultPartition = createResultPartition(ResultPartitionType.BLOCKING, 1); + assertTrue(resultPartition instanceof SortMergeResultPartition); + } + + @Test + public void testReleaseOnConsumptionForPipelinedPartition() { final ResultPartition resultPartition = createResultPartition(ResultPartitionType.PIPELINED); resultPartition.onConsumedSubpartition(0); @@ -81,7 +87,7 @@ public class ResultPartitionFactoryTest extends TestLogger { } @Test - public void testNoConsumptionOnReleaseForBlocking() { + public void testNoReleaseOnConsumptionForBoundedBlockingPartition() { final ResultPartition resultPartition = createResultPartition(ResultPartitionType.BLOCKING); resultPartition.onConsumedSubpartition(0); @@ -89,7 +95,22 @@ public class ResultPartitionFactoryTest extends TestLogger { assertFalse(resultPartition.isReleased()); } + @Test + public void testNoReleaseOnConsumptionForSortMergePartition() { + final ResultPartition resultPartition = createResultPartition(ResultPartitionType.BLOCKING, 1); + + resultPartition.onConsumedSubpartition(0); + + assertFalse(resultPartition.isReleased()); + } + private static ResultPartition createResultPartition(ResultPartitionType partitionType) { + return createResultPartition(partitionType, Integer.MAX_VALUE); + } + + private static ResultPartition createResultPartition( + ResultPartitionType partitionType, + int sortShuffleMinParallelism) { final ResultPartitionManager manager = new ResultPartitionManager(); final ResultPartitionFactory factory = new ResultPartitionFactory( @@ -102,7 +123,9 @@ public class ResultPartitionFactoryTest extends TestLogger { SEGMENT_SIZE, false, "LZ4", - Integer.MAX_VALUE); + Integer.MAX_VALUE, + 10, + sortShuffleMinParallelism); final ResultPartitionDeploymentDescriptor descriptor = new ResultPartitionDeploymentDescriptor( PartitionDescriptorBuilder diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/BlockingShuffleITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/BlockingShuffleITCase.java new file mode 100644 index 0000000..74c3aef --- /dev/null +++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/BlockingShuffleITCase.java @@ -0,0 +1,123 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.runtime; + +import org.apache.flink.api.common.functions.MapFunction; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.NettyShuffleEnvironmentOptions; +import org.apache.flink.runtime.jobgraph.JobGraph; +import org.apache.flink.runtime.jobgraph.ScheduleMode; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.functions.sink.SinkFunction; +import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction; +import org.apache.flink.streaming.api.graph.GlobalDataExchangeMode; +import org.apache.flink.streaming.api.graph.StreamGraph; +import org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator; + +import org.junit.Test; + +import static org.junit.Assert.assertEquals; + +/** + * Tests for blocking shuffle. + */ +public class BlockingShuffleITCase { + + private static final String RECORD = "hello, world!"; + + private final int numTaskManagers = 2; + + private final int numSlotsPerTaskManager = 4; + + @Test + public void testBoundedBlockingShuffle() throws Exception { + JobGraph jobGraph = createJobGraph(1000000); + Configuration configuration = new Configuration(); + JobGraphRunningUtil.execute(jobGraph, configuration, numTaskManagers, numSlotsPerTaskManager); + } + + @Test + public void testBoundedBlockingShuffleWithoutData() throws Exception { + JobGraph jobGraph = createJobGraph(0); + Configuration configuration = new Configuration(); + JobGraphRunningUtil.execute(jobGraph, configuration, numTaskManagers, numSlotsPerTaskManager); + } + + @Test + public void testSortMergeBlockingShuffle() throws Exception { + Configuration configuration = new Configuration(); + configuration.setInteger(NettyShuffleEnvironmentOptions.NETWORK_SORT_SHUFFLE_MIN_PARALLELISM, 1); + + JobGraph jobGraph = createJobGraph(1000000); + JobGraphRunningUtil.execute(jobGraph, configuration, numTaskManagers, numSlotsPerTaskManager); + } + + @Test + public void testSortMergeBlockingShuffleWithoutData() throws Exception { + Configuration configuration = new Configuration(); + configuration.setInteger(NettyShuffleEnvironmentOptions.NETWORK_SORT_SHUFFLE_MIN_PARALLELISM, 1); + + JobGraph jobGraph = createJobGraph(0); + JobGraphRunningUtil.execute(jobGraph, configuration, numTaskManagers, numSlotsPerTaskManager); + } + + private JobGraph createJobGraph(int numRecordsToSend) { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.setParallelism(numTaskManagers * numSlotsPerTaskManager); + DataStream<String> source = env.addSource(new StringSource(numRecordsToSend)); + source + .rebalance().map((MapFunction<String, String>) value -> value) + .broadcast().addSink(new VerifySink()); + + StreamGraph streamGraph = env.getStreamGraph(); + streamGraph.setGlobalDataExchangeMode(GlobalDataExchangeMode.ALL_EDGES_BLOCKING); + streamGraph.setScheduleMode(ScheduleMode.LAZY_FROM_SOURCES); + return StreamingJobGraphGenerator.createJobGraph(streamGraph); + } + + private static class StringSource implements ParallelSourceFunction<String> { + private volatile boolean isRunning = true; + private int numRecordsToSend; + + StringSource(int numRecordsToSend) { + this.numRecordsToSend = numRecordsToSend; + } + + @Override + public void run(SourceContext<String> ctx) throws Exception { + while (isRunning && numRecordsToSend-- > 0) { + ctx.collect(RECORD); + } + } + + @Override + public void cancel() { + isRunning = false; + } + } + + private static class VerifySink implements SinkFunction<String> { + + @Override + public void invoke(String value) throws Exception { + assertEquals(RECORD, value); + } + } +} diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/JobGraphRunningUtil.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/JobGraphRunningUtil.java new file mode 100644 index 0000000..f2c93a3 --- /dev/null +++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/JobGraphRunningUtil.java @@ -0,0 +1,62 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.test.runtime; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.client.program.MiniClusterClient; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.MemorySize; +import org.apache.flink.configuration.TaskManagerOptions; +import org.apache.flink.runtime.jobgraph.JobGraph; +import org.apache.flink.runtime.jobmaster.JobResult; +import org.apache.flink.runtime.minicluster.MiniCluster; +import org.apache.flink.runtime.minicluster.MiniClusterConfiguration; + +/** + * Utils to run {@link JobGraph} on {@link MiniCluster}. + */ +public class JobGraphRunningUtil { + + public static void execute( + JobGraph jobGraph, + Configuration configuration, + int numTaskManagers, + int numSlotsPerTaskManager) throws Exception { + configuration.set(TaskManagerOptions.TOTAL_FLINK_MEMORY, MemorySize.parse("1g")); + + final MiniClusterConfiguration miniClusterConfiguration = new MiniClusterConfiguration.Builder() + .setConfiguration(configuration) + .setNumTaskManagers(numTaskManagers) + .setNumSlotsPerTaskManager(numSlotsPerTaskManager) + .build(); + + try (MiniCluster miniCluster = new MiniCluster(miniClusterConfiguration)) { + miniCluster.start(); + + MiniClusterClient miniClusterClient = new MiniClusterClient(configuration, miniCluster); + // wait for the submission to succeed + JobID jobID = miniClusterClient.submitJob(jobGraph).get(); + + JobResult jobResult = miniClusterClient.requestJobResult(jobID).get(); + if (jobResult.getSerializedThrowable().isPresent()) { + throw new AssertionError(jobResult.getSerializedThrowable().get()); + } + } + } +} diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/ShuffleCompressionITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/ShuffleCompressionITCase.java index 3e118a2..ee1db78 100644 --- a/flink-tests/src/test/java/org/apache/flink/test/runtime/ShuffleCompressionITCase.java +++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/ShuffleCompressionITCase.java @@ -20,12 +20,8 @@ package org.apache.flink.test.runtime; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.ExecutionMode; -import org.apache.flink.api.common.JobID; -import org.apache.flink.client.program.MiniClusterClient; import org.apache.flink.configuration.Configuration; -import org.apache.flink.configuration.MemorySize; import org.apache.flink.configuration.NettyShuffleEnvironmentOptions; -import org.apache.flink.configuration.TaskManagerOptions; import org.apache.flink.runtime.execution.Environment; import org.apache.flink.runtime.io.network.api.reader.MutableRecordReader; import org.apache.flink.runtime.io.network.api.writer.RecordWriter; @@ -38,9 +34,6 @@ import org.apache.flink.runtime.jobgraph.JobVertex; import org.apache.flink.runtime.jobgraph.ScheduleMode; import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable; import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup; -import org.apache.flink.runtime.jobmaster.JobResult; -import org.apache.flink.runtime.minicluster.MiniCluster; -import org.apache.flink.runtime.minicluster.MiniClusterConfiguration; import org.apache.flink.runtime.util.EnvironmentInformation; import org.apache.flink.streaming.runtime.partitioner.BroadcastPartitioner; import org.apache.flink.types.LongValue; @@ -86,32 +79,12 @@ public class ShuffleCompressionITCase { @Test public void testDataCompressionForBlockingShuffle() throws Exception { - executeTest(createJobGraph(ScheduleMode.LAZY_FROM_SOURCES, ResultPartitionType.BLOCKING, ExecutionMode.BATCH)); - } - - private void executeTest(JobGraph jobGraph) throws Exception { Configuration configuration = new Configuration(); - configuration.set(TaskManagerOptions.TOTAL_FLINK_MEMORY, MemorySize.parse("1g")); configuration.setBoolean(NettyShuffleEnvironmentOptions.BLOCKING_SHUFFLE_COMPRESSION_ENABLED, true); - final MiniClusterConfiguration miniClusterConfiguration = new MiniClusterConfiguration.Builder() - .setConfiguration(configuration) - .setNumTaskManagers(NUM_TASKMANAGERS) - .setNumSlotsPerTaskManager(NUM_SLOTS) - .build(); - - try (MiniCluster miniCluster = new MiniCluster(miniClusterConfiguration)) { - miniCluster.start(); - - MiniClusterClient miniClusterClient = new MiniClusterClient(configuration, miniCluster); - // wait for the submission to succeed - JobID jobID = miniClusterClient.submitJob(jobGraph).get(); - - JobResult jobResult = miniClusterClient.requestJobResult(jobID).get(); - if (jobResult.getSerializedThrowable().isPresent()) { - throw new AssertionError(jobResult.getSerializedThrowable().get()); - } - } + JobGraph jobGraph = createJobGraph( + ScheduleMode.LAZY_FROM_SOURCES, ResultPartitionType.BLOCKING, ExecutionMode.BATCH); + JobGraphRunningUtil.execute(jobGraph, configuration, NUM_TASKMANAGERS, NUM_SLOTS); } private static JobGraph createJobGraph(