You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "WencongLiu (via GitHub)" <gi...@apache.org> on 2023/07/10 12:51:37 UTC
[GitHub] [flink] WencongLiu opened a new pull request, #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
WencongLiu opened a new pull request, #22975:
URL: https://github.com/apache/flink/pull/22975
## What is the purpose of the change
*Introduce the configurations for tiered storage*
## Brief change log
- *Introduce the configurations for tiered storage.*
## 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? (no)
--
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 #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #22975:
URL: https://github.com/apache/flink/pull/22975#issuecomment-1628902190
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3c34fdf736b934951415b6282ee7cbb69f205e4f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3c34fdf736b934951415b6282ee7cbb69f205e4f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3c34fdf736b934951415b6282ee7cbb69f205e4f 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] WencongLiu commented on pull request #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "WencongLiu (via GitHub)" <gi...@apache.org>.
WencongLiu commented on PR #22975:
URL: https://github.com/apache/flink/pull/22975#issuecomment-1646580941
@flinkbot run azure
--
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] WencongLiu commented on pull request #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "WencongLiu (via GitHub)" <gi...@apache.org>.
WencongLiu commented on PR #22975:
URL: https://github.com/apache/flink/pull/22975#issuecomment-1645155920
Thanks for the review 😄 @TanYuxin-tyx @reswqa . Please take a look.
--
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 #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #22975:
URL: https://github.com/apache/flink/pull/22975#discussion_r1270166549
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java:
##########
@@ -122,6 +155,9 @@ public ResultPartitionFactory(
this.hybridShuffleSpilledIndexRegionGroupSize = hybridShuffleSpilledIndexRegionGroupSize;
this.hybridShuffleNumRetainedInMemoryRegionsMax =
hybridShuffleNumRetainedInMemoryRegionsMax;
+ this.tieredStorageConfiguration = tieredStorageConfiguration;
+ this.tieredStorageNettyService = tieredStorageNettyService;
+ this.tieredStorageResourceRegistry = tieredStorageResourceRegistry;
Review Comment:
I wonder can we combine all stuff related to `tieredStorage` to a single container class and pass a optional<That Class> to this `ResultPartitionFactory`.
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -28,6 +28,12 @@
@PublicEvolving
public class NettyShuffleEnvironmentOptions {
+ private static final String HYBRID_SHUFFLE_NEW_MODE_OPTION_NAME =
+ "taskmanager.network.hybrid-shuffle.enable-new-mode";
+
+ private static final String HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH_OPTION_NAME =
+ "taskmanager.network.hybrid-shuffle.remote.path";
Review Comment:
It seems strange that the other options do not define key name separately, so I suggest maintaining consistency.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java:
##########
@@ -18,58 +18,435 @@
package org.apache.flink.runtime.io.network.partition.hybrid.tiered.common;
-import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.BufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.HashBufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.SortBufferAccumulator;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.disk.DiskIOScheduler;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.disk.DiskTierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.memory.MemoryTierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.remote.RemoteTierFactory;
-import javax.annotation.Nullable;
-
-import java.util.Arrays;
+import java.time.Duration;
+import java.util.ArrayList;
import java.util.List;
/** Configurations for the Tiered Storage. */
public class TieredStorageConfiguration {
- // TODO, after implementing the tier factory, add appreciate implementations to the array.
- private static final TierFactory[] DEFAULT_MEMORY_DISK_TIER_FACTORIES = new TierFactory[0];
+ private static final String DEFAULT_REMOTE_STORAGE_BASE_PATH = null;
+
+ private static final int DEFAULT_TIERED_STORAGE_BUFFER_SIZE = 32 * 1024;
+
+ private static final int DEFAULT_MEMORY_TIER_EXCLUSIVE_BUFFERS = 100;
+
+ private static final int DEFAULT_DISK_TIER_EXCLUSIVE_BUFFERS = 1;
+
+ private static final int DEFAULT_REMOTE_TIER_EXCLUSIVE_BUFFERS = 1;
+
+ private static final int DEFAULT_NUM_BUFFERS_USE_SORT_ACCUMULATOR_THRESHOLD = 512;
+
+ private static final int DEFAULT_MEMORY_TIER_NUM_BYTES_PER_SEGMENT = 320 * 1024;
+
+ private static final int DEFAULT_DISK_TIER_NUM_BYTES_PER_SEGMENT = 8 * 1024 * 1024;
+
+ private static final int DEFAULT_REMOTE_TIER_NUM_BYTES_PER_SEGMENT = 8 * 1024 * 1024;
+
+ private static final float DEFAULT_NUM_BUFFERS_TRIGGER_FLUSH_RATIO = 0.5f;
+
+ private static final int DEFAULT_DISK_TIER_MAX_BUFFERS_READ_AHEAD = 5;
+
+ private static final Duration DEFAULT_DISK_TIER_BUFFER_REQUEST_TIMEOUT = Duration.ofMinutes(5);
+
+ private static final float DEFAULT_MIN_RESERVE_SPACE_FRACTION = 0.05f;
+
+ private static final int DEFAULT_REGION_GROUP_SIZE_IN_BYTES = 1024;
+
+ private static final long DEFAULT_MAX_REGION_NUM_RETAINED_IN_MEMORY = 1024 * 1024L;
+
+ private final String remoteStorageBasePath;
- /** If the remote storage tier is not used, this field may be null. */
- @Nullable private final String remoteStorageBasePath;
+ private final int tieredStorageBufferSize;
- public TieredStorageConfiguration(@Nullable String remoteStorageBasePath) {
+ private final int memoryTierExclusiveBuffers;
+
+ private final int diskTierExclusiveBuffers;
+
+ private final int remoteTierExclusiveBuffers;
+
+ private final int accumulatorExclusiveBuffers;
+
+ private final int memoryTierNumBytesPerSegment;
+
+ private final int diskTierNumBytesPerSegment;
+
+ private final int remoteTierNumBytesPerSegment;
+
+ private final float numBuffersTriggerFlushRatio;
+
+ private final int diskIOSchedulerMaxBuffersReadAhead;
+
+ private final Duration diskIOSchedulerRequestTimeout;
+
+ private final float minReserveSpaceFraction;
Review Comment:
```suggestion
private final float minReserveDiskSpaceFraction;
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/NettyShuffleUtilsTest.java:
##########
@@ -99,6 +100,8 @@ public void testComputeRequiredNetworkBuffers() throws Exception {
maxRequiredBuffersPerGate,
sortShuffleMinParallelism,
numSortShuffleMinBuffers,
+ false,
+ tieredStoreExclusiveBuffers,
Review Comment:
The last two commits need to be tested, especially the `GateBuffersSpecTest`.
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -381,6 +393,38 @@ public class NettyShuffleEnvironmentOptions {
+ "tasks have occupied all the buffers and the downstream tasks are waiting for the exclusive buffers. The timeout breaks"
+ "the tie by failing the request of exclusive buffers and ask users to increase the number of total buffers.");
+ /** The option to enable the new mode of hybrid shuffle. */
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ public static final ConfigOption<Boolean> NETWORK_HYBRID_SHUFFLE_ENABLE_NEW_MODE =
+ ConfigOptions.key(HYBRID_SHUFFLE_NEW_MODE_OPTION_NAME)
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "The option is used to enable the new mode of hybrid shuffle, which has resolved existing issues in the legacy mode. First, the new mode "
+ + "uses less required network memory. Second, the new mode can store shuffle data in remote storage when the disk space is not "
+ + "enough, which could avoid insufficient disk space errors and is only supported when "
+ + HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH_OPTION_NAME
+ + " is configured. The new mode is currently in an experimental phase. The option can be set to false to fallback to the legacy "
+ + "mode if the new mode is unstable. Once the new mode reaches a stable state, the legacy mode and the option will be removed.");
Review Comment:
```suggestion
Once the new mode reaches a stable state, the legacy mode as well as this option will be removed.");
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleMaster.java:
##########
@@ -67,6 +75,17 @@ public NettyShuffleMaster(Configuration conf) {
conf.getInteger(NettyShuffleEnvironmentOptions.NETWORK_SORT_SHUFFLE_MIN_BUFFERS);
networkBufferSize = ConfigurationParserUtils.getPageSize(conf);
+ if (conf.getBoolean(NETWORK_HYBRID_SHUFFLE_ENABLE_NEW_MODE)) {
+ tieredStorageConfiguration =
+ TieredStorageConfiguration.builder(
+ conf.getString(NETWORK_HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH))
+ .build();
+ tieredInternalShuffleMaster = new TieredInternalShuffleMaster(conf);
+ } else {
+ tieredStorageConfiguration = null;
+ tieredInternalShuffleMaster = null;
+ }
Review Comment:
This is redundant.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java:
##########
@@ -18,58 +18,435 @@
package org.apache.flink.runtime.io.network.partition.hybrid.tiered.common;
-import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.BufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.HashBufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.SortBufferAccumulator;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.disk.DiskIOScheduler;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.disk.DiskTierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.memory.MemoryTierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.remote.RemoteTierFactory;
-import javax.annotation.Nullable;
-
-import java.util.Arrays;
+import java.time.Duration;
+import java.util.ArrayList;
import java.util.List;
/** Configurations for the Tiered Storage. */
public class TieredStorageConfiguration {
Review Comment:
The second commit `Enable the TieredStorageConfiguration` is a bit unclear, I suggest that rename this to `Enable TieredStorageResultPartition` and squash it with the first commit.
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -381,6 +393,38 @@ public class NettyShuffleEnvironmentOptions {
+ "tasks have occupied all the buffers and the downstream tasks are waiting for the exclusive buffers. The timeout breaks"
+ "the tie by failing the request of exclusive buffers and ask users to increase the number of total buffers.");
+ /** The option to enable the new mode of hybrid shuffle. */
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ public static final ConfigOption<Boolean> NETWORK_HYBRID_SHUFFLE_ENABLE_NEW_MODE =
+ ConfigOptions.key(HYBRID_SHUFFLE_NEW_MODE_OPTION_NAME)
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "The option is used to enable the new mode of hybrid shuffle, which has resolved existing issues in the legacy mode. First, the new mode "
+ + "uses less required network memory. Second, the new mode can store shuffle data in remote storage when the disk space is not "
+ + "enough, which could avoid insufficient disk space errors and is only supported when "
+ + HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH_OPTION_NAME
+ + " is configured. The new mode is currently in an experimental phase. The option can be set to false to fallback to the legacy "
Review Comment:
```suggestion
+ " is configured. The new mode is currently in an experimental phase. It can be set to false to fallback to the legacy mode if something unexpected."
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java:
##########
@@ -164,7 +164,10 @@ private static ResultPartition createResultPartition(
false,
0,
256,
- Long.MAX_VALUE);
+ Long.MAX_VALUE,
Review Comment:
We does need some test for tieredStorage.
--
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] TanYuxin-tyx commented on a diff in pull request #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22975:
URL: https://github.com/apache/flink/pull/22975#discussion_r1270157473
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java:
##########
@@ -227,23 +263,78 @@ public ResultPartition create(
}
} else if (type == ResultPartitionType.HYBRID_FULL
|| type == ResultPartitionType.HYBRID_SELECTIVE) {
- partition =
- new HsResultPartition(
- taskNameWithSubtaskAndId,
- partitionIndex,
- id,
- type,
- subpartitions.length,
- maxParallelism,
- batchShuffleReadBufferPool,
- batchShuffleReadIOExecutor,
- partitionManager,
- channelManager.createChannel().getPath(),
- networkBufferSize,
- getHybridShuffleConfiguration(numberOfSubpartitions, type),
- bufferCompressor,
- isBroadcast,
- bufferPoolFactory);
+ if (tieredStorageConfiguration != null) {
+ // Create memory manager.
+ TieredStorageMemoryManager memoryManager =
+ new TieredStorageMemoryManagerImpl(
+ checkNotNull(tieredStorageConfiguration)
+ .getNumBuffersTriggerFlushRatio(),
+ true);
+
+ // Create buffer accumulator.
+ int accumulatorExclusiveBufferNum =
Review Comment:
Maybe use `numAccumulatorExclusiveBuffers` to replace `accumulatorExclusiveBufferNum`.
--
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 #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #22975:
URL: https://github.com/apache/flink/pull/22975#discussion_r1270333895
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -28,6 +28,12 @@
@PublicEvolving
public class NettyShuffleEnvironmentOptions {
+ private static final String HYBRID_SHUFFLE_NEW_MODE_OPTION_NAME =
+ "taskmanager.network.hybrid-shuffle.enable-new-mode";
+
+ private static final String HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH_OPTION_NAME =
+ "taskmanager.network.hybrid-shuffle.remote.path";
Review Comment:
Fair enough.
--
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] WencongLiu commented on a diff in pull request #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "WencongLiu (via GitHub)" <gi...@apache.org>.
WencongLiu commented on code in PR #22975:
URL: https://github.com/apache/flink/pull/22975#discussion_r1270307929
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -381,6 +393,38 @@ public class NettyShuffleEnvironmentOptions {
+ "tasks have occupied all the buffers and the downstream tasks are waiting for the exclusive buffers. The timeout breaks"
+ "the tie by failing the request of exclusive buffers and ask users to increase the number of total buffers.");
+ /** The option to enable the new mode of hybrid shuffle. */
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ public static final ConfigOption<Boolean> NETWORK_HYBRID_SHUFFLE_ENABLE_NEW_MODE =
+ ConfigOptions.key(HYBRID_SHUFFLE_NEW_MODE_OPTION_NAME)
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "The option is used to enable the new mode of hybrid shuffle, which has resolved existing issues in the legacy mode. First, the new mode "
+ + "uses less required network memory. Second, the new mode can store shuffle data in remote storage when the disk space is not "
+ + "enough, which could avoid insufficient disk space errors and is only supported when "
+ + HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH_OPTION_NAME
+ + " is configured. The new mode is currently in an experimental phase. The option can be set to false to fallback to the legacy "
+ + "mode if the new mode is unstable. Once the new mode reaches a stable state, the legacy mode and the option will be removed.");
Review Comment:
Fixed.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java:
##########
@@ -18,58 +18,435 @@
package org.apache.flink.runtime.io.network.partition.hybrid.tiered.common;
-import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.BufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.HashBufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.SortBufferAccumulator;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.disk.DiskIOScheduler;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.disk.DiskTierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.memory.MemoryTierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.remote.RemoteTierFactory;
-import javax.annotation.Nullable;
-
-import java.util.Arrays;
+import java.time.Duration;
+import java.util.ArrayList;
import java.util.List;
/** Configurations for the Tiered Storage. */
public class TieredStorageConfiguration {
- // TODO, after implementing the tier factory, add appreciate implementations to the array.
- private static final TierFactory[] DEFAULT_MEMORY_DISK_TIER_FACTORIES = new TierFactory[0];
+ private static final String DEFAULT_REMOTE_STORAGE_BASE_PATH = null;
+
+ private static final int DEFAULT_TIERED_STORAGE_BUFFER_SIZE = 32 * 1024;
+
+ private static final int DEFAULT_MEMORY_TIER_EXCLUSIVE_BUFFERS = 100;
+
+ private static final int DEFAULT_DISK_TIER_EXCLUSIVE_BUFFERS = 1;
+
+ private static final int DEFAULT_REMOTE_TIER_EXCLUSIVE_BUFFERS = 1;
+
+ private static final int DEFAULT_NUM_BUFFERS_USE_SORT_ACCUMULATOR_THRESHOLD = 512;
+
+ private static final int DEFAULT_MEMORY_TIER_NUM_BYTES_PER_SEGMENT = 320 * 1024;
+
+ private static final int DEFAULT_DISK_TIER_NUM_BYTES_PER_SEGMENT = 8 * 1024 * 1024;
+
+ private static final int DEFAULT_REMOTE_TIER_NUM_BYTES_PER_SEGMENT = 8 * 1024 * 1024;
+
+ private static final float DEFAULT_NUM_BUFFERS_TRIGGER_FLUSH_RATIO = 0.5f;
+
+ private static final int DEFAULT_DISK_TIER_MAX_BUFFERS_READ_AHEAD = 5;
+
+ private static final Duration DEFAULT_DISK_TIER_BUFFER_REQUEST_TIMEOUT = Duration.ofMinutes(5);
+
+ private static final float DEFAULT_MIN_RESERVE_SPACE_FRACTION = 0.05f;
+
+ private static final int DEFAULT_REGION_GROUP_SIZE_IN_BYTES = 1024;
+
+ private static final long DEFAULT_MAX_REGION_NUM_RETAINED_IN_MEMORY = 1024 * 1024L;
+
+ private final String remoteStorageBasePath;
- /** If the remote storage tier is not used, this field may be null. */
- @Nullable private final String remoteStorageBasePath;
+ private final int tieredStorageBufferSize;
- public TieredStorageConfiguration(@Nullable String remoteStorageBasePath) {
+ private final int memoryTierExclusiveBuffers;
+
+ private final int diskTierExclusiveBuffers;
+
+ private final int remoteTierExclusiveBuffers;
+
+ private final int accumulatorExclusiveBuffers;
+
+ private final int memoryTierNumBytesPerSegment;
+
+ private final int diskTierNumBytesPerSegment;
+
+ private final int remoteTierNumBytesPerSegment;
+
+ private final float numBuffersTriggerFlushRatio;
+
+ private final int diskIOSchedulerMaxBuffersReadAhead;
+
+ private final Duration diskIOSchedulerRequestTimeout;
+
+ private final float minReserveSpaceFraction;
Review Comment:
Fixed.
--
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] TanYuxin-tyx commented on a diff in pull request #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22975:
URL: https://github.com/apache/flink/pull/22975#discussion_r1269371029
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java:
##########
@@ -83,6 +106,13 @@ public class ResultPartitionFactory {
private final int maxOverdraftBuffersPerGate;
+ // The following attributes will be null if tiered storage shuffle is disabled.
Review Comment:
Use /* */ doc instead here.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java:
##########
@@ -101,9 +101,10 @@ public class SingleInputGateFactory {
private final BufferDebloatConfiguration debloatConfiguration;
- @Nullable
- private final TieredStorageConfiguration
- tieredStorageConfiguration; // is null if tiered storage shuffle is disabled.
+ // The following attributes will be null if tiered storage shuffle is disabled.
Review Comment:
Use `/* */` instead.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java:
##########
@@ -266,6 +357,74 @@ private HybridShuffleConfiguration getHybridShuffleConfiguration(
.build();
}
+ private BufferAccumulator createBufferAccumulator(
+ int numSubpartitions,
+ int accumulatorExclusiveBufferNum,
Review Comment:
ditto, `numAccumulatorExclusiveBuffers`
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java:
##########
@@ -227,23 +263,78 @@ public ResultPartition create(
}
} else if (type == ResultPartitionType.HYBRID_FULL
|| type == ResultPartitionType.HYBRID_SELECTIVE) {
- partition =
- new HsResultPartition(
- taskNameWithSubtaskAndId,
- partitionIndex,
- id,
- type,
- subpartitions.length,
- maxParallelism,
- batchShuffleReadBufferPool,
- batchShuffleReadIOExecutor,
- partitionManager,
- channelManager.createChannel().getPath(),
- networkBufferSize,
- getHybridShuffleConfiguration(numberOfSubpartitions, type),
- bufferCompressor,
- isBroadcast,
- bufferPoolFactory);
+ if (tieredStorageConfiguration != null) {
+ // Create memory manager.
+ TieredStorageMemoryManager memoryManager =
+ new TieredStorageMemoryManagerImpl(
+ checkNotNull(tieredStorageConfiguration)
+ .getNumBuffersTriggerFlushRatio(),
+ true);
+
+ // Create buffer accumulator.
+ int accumulatorExclusiveBufferNum =
Review Comment:
Maybe use `numAccumulatorExclusiveBuffers` to replace `accumulatorExclusiveBufferNum`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java:
##########
@@ -18,58 +18,435 @@
package org.apache.flink.runtime.io.network.partition.hybrid.tiered.common;
-import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.BufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.HashBufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.SortBufferAccumulator;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.disk.DiskIOScheduler;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.disk.DiskTierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.memory.MemoryTierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.remote.RemoteTierFactory;
-import javax.annotation.Nullable;
-
-import java.util.Arrays;
+import java.time.Duration;
+import java.util.ArrayList;
import java.util.List;
/** Configurations for the Tiered Storage. */
public class TieredStorageConfiguration {
- // TODO, after implementing the tier factory, add appreciate implementations to the array.
- private static final TierFactory[] DEFAULT_MEMORY_DISK_TIER_FACTORIES = new TierFactory[0];
+ private static final String DEFAULT_REMOTE_STORAGE_BASE_PATH = null;
+
+ private static final int DEFAULT_TIERED_STORAGE_BUFFER_SIZE = 32 * 1024;
+
+ private static final int DEFAULT_MEMORY_TIER_EXCLUSIVE_BUFFERS = 100;
+
+ private static final int DEFAULT_DISK_TIER_EXCLUSIVE_BUFFERS = 1;
+
+ private static final int DEFAULT_REMOTE_TIER_EXCLUSIVE_BUFFERS = 1;
+
+ private static final int DEFAULT_NUM_BUFFERS_USE_SORT_ACCUMULATOR_THRESHOLD = 512;
+
+ private static final int DEFAULT_MEMORY_TIER_NUM_BYTES_PER_SEGMENT = 320 * 1024;
+
+ private static final int DEFAULT_DISK_TIER_NUM_BYTES_PER_SEGMENT = 8 * 1024 * 1024;
+
+ private static final int DEFAULT_REMOTE_TIER_NUM_BYTES_PER_SEGMENT = 8 * 1024 * 1024;
+
+ private static final float DEFAULT_NUM_BUFFERS_TRIGGER_FLUSH_RATIO = 0.5f;
Review Comment:
Here I have a concern.
Maybe 0.5 is too low to trigger the flush process, we'd better change to at least 0.6 or more. Because too less a trigger ratio will lead to generating too much little size regions, which is bad for sequential reading performance.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java:
##########
@@ -227,23 +263,78 @@ public ResultPartition create(
}
} else if (type == ResultPartitionType.HYBRID_FULL
|| type == ResultPartitionType.HYBRID_SELECTIVE) {
- partition =
- new HsResultPartition(
- taskNameWithSubtaskAndId,
- partitionIndex,
- id,
- type,
- subpartitions.length,
- maxParallelism,
- batchShuffleReadBufferPool,
- batchShuffleReadIOExecutor,
- partitionManager,
- channelManager.createChannel().getPath(),
- networkBufferSize,
- getHybridShuffleConfiguration(numberOfSubpartitions, type),
- bufferCompressor,
- isBroadcast,
- bufferPoolFactory);
+ if (tieredStorageConfiguration != null) {
+ // Create memory manager.
+ TieredStorageMemoryManager memoryManager =
+ new TieredStorageMemoryManagerImpl(
+ checkNotNull(tieredStorageConfiguration)
+ .getNumBuffersTriggerFlushRatio(),
+ true);
+
+ // Create buffer accumulator.
+ int accumulatorExclusiveBufferNum =
Review Comment:
Maybe use `numAccumulatorExclusiveBuffers` to replace `accumulatorExclusiveBufferNum`.
--
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 #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on PR #22975:
URL: https://github.com/apache/flink/pull/22975#issuecomment-1646880017
Finally, CI passed 🤣, merging...
--
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 #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #22975:
URL: https://github.com/apache/flink/pull/22975#discussion_r1265245023
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -379,6 +391,38 @@ public class NettyShuffleEnvironmentOptions {
+ "tasks have occupied all the buffers and the downstream tasks are waiting for the exclusive buffers. The timeout breaks"
+ "the tie by failing the request of exclusive buffers and ask users to increase the number of total buffers.");
+ /** The option to enable the new mode of hybrid shuffle. */
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ public static final ConfigOption<Boolean> NETWORK_HYBRID_SHUFFLE_ENABLE_NEW_MODE =
+ ConfigOptions.key(HYBRID_SHUFFLE_NEW_MODE_OPTION_NAME)
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "The option is used to enable the new mode of hybrid shuffle, which has resolved existing issues in the legacy mode. First, the new mode "
+ + "uses less required network memory. Second, the new mode can store shuffle data in remote storage when the disk space is not "
+ + "enough, which could avoid insufficient disk space errors and is only supported when "
+ + HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH_OPTION_NAME
+ + " is configured. The new mode is currently in an experimental phase. The option can be set to false to fallback to the legacy "
+ + "mode if the new mode is unstable. Once the new mode reaches a stable state, the option will be removed.");
Review Comment:
```suggestion
+ "mode if the new mode is unstable. Once the new mode reaches a stable state, the legacy mode and the option will be 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 #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #22975:
URL: https://github.com/apache/flink/pull/22975#discussion_r1259626675
##########
flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleMaster.java:
##########
@@ -49,7 +55,9 @@ public class NettyShuffleMaster implements ShuffleMaster<NettyShuffleDescriptor>
private final int networkBufferSize;
- // TODO, WIP: create tiered internal shuffle master after enabling the tiered storage.
+ @Nullable private TieredInternalShuffleMaster tieredInternalShuffleMaster;
+
+ @Nullable private TieredStorageConfiguration tieredStorageConfiguration;
Review Comment:
Should be final.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredInternalShuffleMaster.java:
##########
@@ -40,7 +41,9 @@ public class TieredInternalShuffleMaster {
public TieredInternalShuffleMaster(Configuration conf) {
TieredStorageConfiguration tieredStorageConfiguration =
- TieredStorageConfiguration.fromConfiguration(conf);
+ TieredStorageConfiguration.builder(
+ conf.getString(NETWORK_HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH))
+ .build();
Review Comment:
This change seems belonging to the first commit. Otherwise, there will be a compiling error.
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -379,6 +380,28 @@ public class NettyShuffleEnvironmentOptions {
+ "tasks have occupied all the buffers and the downstream tasks are waiting for the exclusive buffers. The timeout breaks"
+ "the tie by failing the request of exclusive buffers and ask users to increase the number of total buffers.");
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ @Internal
+ public static final ConfigOption<Boolean> NETWORK_HYBRID_SHUFFLE_ENABLE_TIERED_STORAGE =
+ ConfigOptions.key("taskmanager.network.hybrid-shuffle.enable-tiered-storage")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "The option is used to enable tiered storage architecture for hybrid shuffle mode.");
+
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ @Internal
+ public static final ConfigOption<String> NETWORK_HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH =
+ key("taskmanager.network.hybrid-shuffle.remote.path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The base home path of remote storage for remote tier. If the option is configured, "
+ + "Hybrid Shuffle will use the remote storage path as a supplement to the"
+ + "local disks. If not configured, the remote storage will not be used.");
Review Comment:
1. This is not internal
2. JavaDoc is missing
3. Shall we expose the concept `remote tier` to users at all? What are the minimum necessary things that user needs to understand?
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -379,6 +380,28 @@ public class NettyShuffleEnvironmentOptions {
+ "tasks have occupied all the buffers and the downstream tasks are waiting for the exclusive buffers. The timeout breaks"
+ "the tie by failing the request of exclusive buffers and ask users to increase the number of total buffers.");
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ @Internal
+ public static final ConfigOption<Boolean> NETWORK_HYBRID_SHUFFLE_ENABLE_TIERED_STORAGE =
+ ConfigOptions.key("taskmanager.network.hybrid-shuffle.enable-tiered-storage")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "The option is used to enable tiered storage architecture for hybrid shuffle mode.");
Review Comment:
1. This is not internal.
2. JavaDoc is missing.
3. What is "tiered storage architecture" from the user's perspective? How is it different from the original architecture? What does the user needs to know about this?
4. Why is the default value `false`? How can a user decide which mode to use? Would there be anyone enable it at all?
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -379,6 +380,28 @@ public class NettyShuffleEnvironmentOptions {
+ "tasks have occupied all the buffers and the downstream tasks are waiting for the exclusive buffers. The timeout breaks"
+ "the tie by failing the request of exclusive buffers and ask users to increase the number of total buffers.");
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ @Internal
+ public static final ConfigOption<Boolean> NETWORK_HYBRID_SHUFFLE_ENABLE_TIERED_STORAGE =
+ ConfigOptions.key("taskmanager.network.hybrid-shuffle.enable-tiered-storage")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "The option is used to enable tiered storage architecture for hybrid shuffle mode.");
+
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ @Internal
+ public static final ConfigOption<String> NETWORK_HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH =
+ key("taskmanager.network.hybrid-shuffle.remote.path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The base home path of remote storage for remote tier. If the option is configured, "
+ + "Hybrid Shuffle will use the remote storage path as a supplement to the"
+ + "local disks. If not configured, the remote storage will not be used.");
+
Review Comment:
Is there any config option that only the old mode of hybrid shuffle uses but not the new one?
--
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] WencongLiu commented on pull request #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "WencongLiu (via GitHub)" <gi...@apache.org>.
WencongLiu commented on PR #22975:
URL: https://github.com/apache/flink/pull/22975#issuecomment-1645628610
@flinkbot run azure
--
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] WencongLiu commented on a diff in pull request #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "WencongLiu (via GitHub)" <gi...@apache.org>.
WencongLiu commented on code in PR #22975:
URL: https://github.com/apache/flink/pull/22975#discussion_r1270307421
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java:
##########
@@ -122,6 +155,9 @@ public ResultPartitionFactory(
this.hybridShuffleSpilledIndexRegionGroupSize = hybridShuffleSpilledIndexRegionGroupSize;
this.hybridShuffleNumRetainedInMemoryRegionsMax =
hybridShuffleNumRetainedInMemoryRegionsMax;
+ this.tieredStorageConfiguration = tieredStorageConfiguration;
+ this.tieredStorageNettyService = tieredStorageNettyService;
+ this.tieredStorageResourceRegistry = tieredStorageResourceRegistry;
Review Comment:
I've added a new container class `TieredStorage` to combine these arguments.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleMaster.java:
##########
@@ -67,6 +75,17 @@ public NettyShuffleMaster(Configuration conf) {
conf.getInteger(NettyShuffleEnvironmentOptions.NETWORK_SORT_SHUFFLE_MIN_BUFFERS);
networkBufferSize = ConfigurationParserUtils.getPageSize(conf);
+ if (conf.getBoolean(NETWORK_HYBRID_SHUFFLE_ENABLE_NEW_MODE)) {
+ tieredStorageConfiguration =
+ TieredStorageConfiguration.builder(
+ conf.getString(NETWORK_HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH))
+ .build();
+ tieredInternalShuffleMaster = new TieredInternalShuffleMaster(conf);
+ } else {
+ tieredStorageConfiguration = null;
+ tieredInternalShuffleMaster = null;
+ }
Review Comment:
`tieredStorageConfiguration` and `tieredInternalShuffleMaster` should be `final`.
--
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] WencongLiu commented on a diff in pull request #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "WencongLiu (via GitHub)" <gi...@apache.org>.
WencongLiu commented on code in PR #22975:
URL: https://github.com/apache/flink/pull/22975#discussion_r1270307675
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -28,6 +28,12 @@
@PublicEvolving
public class NettyShuffleEnvironmentOptions {
+ private static final String HYBRID_SHUFFLE_NEW_MODE_OPTION_NAME =
+ "taskmanager.network.hybrid-shuffle.enable-new-mode";
+
+ private static final String HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH_OPTION_NAME =
+ "taskmanager.network.hybrid-shuffle.remote.path";
Review Comment:
The two related static configs need to statically refer the name of each other in the note, so there should be static String names to ensure the random order.
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -381,6 +393,38 @@ public class NettyShuffleEnvironmentOptions {
+ "tasks have occupied all the buffers and the downstream tasks are waiting for the exclusive buffers. The timeout breaks"
+ "the tie by failing the request of exclusive buffers and ask users to increase the number of total buffers.");
+ /** The option to enable the new mode of hybrid shuffle. */
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ public static final ConfigOption<Boolean> NETWORK_HYBRID_SHUFFLE_ENABLE_NEW_MODE =
+ ConfigOptions.key(HYBRID_SHUFFLE_NEW_MODE_OPTION_NAME)
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "The option is used to enable the new mode of hybrid shuffle, which has resolved existing issues in the legacy mode. First, the new mode "
+ + "uses less required network memory. Second, the new mode can store shuffle data in remote storage when the disk space is not "
+ + "enough, which could avoid insufficient disk space errors and is only supported when "
+ + HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH_OPTION_NAME
+ + " is configured. The new mode is currently in an experimental phase. The option can be set to false to fallback to the legacy "
Review Comment:
Fixed.
--
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] TanYuxin-tyx commented on a diff in pull request #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22975:
URL: https://github.com/apache/flink/pull/22975#discussion_r1270396604
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/GateBuffersSpecTest.java:
##########
@@ -143,17 +144,25 @@ void testCalculationWithConfiguredZeroExclusiveBuffer(ResultPartitionType partit
void testConfiguredMaxRequiredBuffersPerGate(ResultPartitionType partitionType) {
Optional<Integer> emptyConfig = Optional.empty();
int effectiveMaxRequiredBuffers =
- getEffectiveMaxRequiredBuffersPerGate(partitionType, emptyConfig);
+ getEffectiveMaxRequiredBuffersPerGate(partitionType, emptyConfig, false);
int expectEffectiveMaxRequiredBuffers =
isPipelinedOrHybridResultPartition(partitionType)
? InputGateSpecUtils.DEFAULT_MAX_REQUIRED_BUFFERS_PER_GATE_FOR_STREAM
- : InputGateSpecUtils.DEFAULT_MAX_REQUIRED_BUFFERS_PER_GATE_FOR_BATCH;
+ : DEFAULT_MAX_REQUIRED_BUFFERS_PER_GATE_FOR_BATCH;
Review Comment:
Why `DEFAULT_MAX_REQUIRED_BUFFERS_PER_GATE_FOR_BATCH` is different with `InputGateSpecUtils.DEFAULT_MAX_REQUIRED_BUFFERS_PER_GATE_FOR_STREAM`.
We'd better keep the two import formats consistent.
--
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 merged pull request #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa merged PR #22975:
URL: https://github.com/apache/flink/pull/22975
--
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 #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #22975:
URL: https://github.com/apache/flink/pull/22975#discussion_r1263432147
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -379,6 +391,40 @@ public class NettyShuffleEnvironmentOptions {
+ "tasks have occupied all the buffers and the downstream tasks are waiting for the exclusive buffers. The timeout breaks"
+ "the tie by failing the request of exclusive buffers and ask users to increase the number of total buffers.");
+ /** The option to enable the new mode of hybrid shuffle. */
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ public static final ConfigOption<Boolean> NETWORK_HYBRID_SHUFFLE_ENABLE_NEW_MODE =
+ ConfigOptions.key(HYBRID_SHUFFLE_NEW_MODE_OPTION_NAME)
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "The option is used to enable the new mode of hybrid shuffle, which has resolved existing issues in the legacy mode. First, the new mode "
+ + "uses less required network memory. For large parallelism jobs, less network memory is required. Second, the new mode "
+ + "can store shuffle data in remote storage when the disk space is not enough, which could avoid insufficient disk space "
+ + "errors and is only supported when "
+ + HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH_OPTION_NAME
+ + " is configured. "
+ + "The option is set true by default, but the new mode can fallback to legacy mode by setting this option false. "
+ + "In the future, the new mode will become the default implementation of hybrid shuffle and this option will be removed.");
Review Comment:
1. "First, the new mode uses less required network memory. For large parallelism jobs, less network memory is required." This sounds duplicated.
2. "but the new mode can fallback to legacy mode by setting this option false." Why would user fallback to the legacy mode?
3. "In the future, the new mode will become the default implementation of hybrid shuffle and this option will be removed." The new mode is already the default implementation.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java:
##########
@@ -18,58 +18,408 @@
package org.apache.flink.runtime.io.network.partition.hybrid.tiered.common;
-import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.BufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.HashBufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.SortBufferAccumulator;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.disk.DiskIOScheduler;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.disk.DiskTierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.memory.MemoryTierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.remote.RemoteTierFactory;
-import javax.annotation.Nullable;
-
-import java.util.Arrays;
+import java.time.Duration;
+import java.util.ArrayList;
import java.util.List;
/** Configurations for the Tiered Storage. */
public class TieredStorageConfiguration {
- // TODO, after implementing the tier factory, add appreciate implementations to the array.
- private static final TierFactory[] DEFAULT_MEMORY_DISK_TIER_FACTORIES = new TierFactory[0];
+ private static final String DEFAULT_REMOTE_STORAGE_BASE_PATH = null;
+
+ private static final int DEFAULT_TIERED_STORAGE_BUFFER_SIZE = 32 * 1024;
+
+ private static final int DEFAULT_MEMORY_TIER_EXCLUSIVE_BUFFERS = 100;
+
+ private static final int DEFAULT_DISK_TIER_EXCLUSIVE_BUFFERS = 1;
+
+ private static final int DEFAULT_REMOTE_TIER_EXCLUSIVE_BUFFERS = 1;
+
+ private static final int DEFAULT_NUM_BUFFERS_USE_SORT_ACCUMULATOR_THRESHOLD = 512;
+
+ private static final int DEFAULT_MEMORY_TIER_NUM_BYTES_PER_SEGMENT = 320 * 1024;
+
+ private static final int DEFAULT_DISK_TIER_NUM_BYTES_PER_SEGMENT = 8 * 1024 * 1024;
+
+ private static final int DEFAULT_REMOTE_TIER_NUM_BYTES_PER_SEGMENT = 8 * 1024 * 1024;
+
+ private static final float DEFAULT_NUM_BUFFERS_TRIGGER_FLUSH_RATIO = 0.5f;
+
+ private static final int DEFAULT_DISK_TIER_MAX_BUFFERS_READ_AHEAD = 5;
+
+ private static final Duration DEFAULT_DISK_TIER_BUFFER_REQUEST_TIMEOUT = Duration.ofMinutes(5);
+
+ private static final float DEFAULT_MIN_RESERVE_SPACE_FRACTION = 0.05f;
+
+ private final String remoteStorageBasePath;
- /** If the remote storage tier is not used, this field may be null. */
- @Nullable private final String remoteStorageBasePath;
+ private final int tieredStorageBufferSize;
- public TieredStorageConfiguration(@Nullable String remoteStorageBasePath) {
+ private final int memoryTierExclusiveBuffers;
+
+ private final int diskTierExclusiveBuffers;
+
+ private final int remoteTierExclusiveBuffers;
+
+ private final int accumulatorExclusiveBuffers;
+
+ private final int memoryTierNumBytesPerSegment;
+
+ private final int diskTierNumBytesPerSegment;
+
+ private final int remoteTierNumBytesPerSegment;
+
+ private final float numBuffersTriggerFlushRatio;
+
+ private final int diskIOSchedulerMaxBuffersReadAhead;
+
+ private final Duration diskIOSchedulerRequestTimeout;
+
+ private final float minReserveSpaceFraction;
+
+ private final List<TierFactory> tierFactories;
+
+ public TieredStorageConfiguration(
+ String remoteStorageBasePath,
+ int tieredStorageBufferSize,
+ int memoryTierExclusiveBuffers,
+ int diskTierExclusiveBuffers,
+ int remoteTierExclusiveBuffers,
+ int accumulatorExclusiveBuffers,
+ int memoryTierNumBytesPerSegment,
+ int diskTierNumBytesPerSegment,
+ int remoteTierNumBytesPerSegment,
+ float numBuffersTriggerFlushRatio,
+ int diskIOSchedulerMaxBuffersReadAhead,
+ Duration diskIOSchedulerRequestTimeout,
+ float minReserveSpaceFraction,
+ List<TierFactory> tierFactories) {
this.remoteStorageBasePath = remoteStorageBasePath;
+ this.tieredStorageBufferSize = tieredStorageBufferSize;
+ this.memoryTierExclusiveBuffers = memoryTierExclusiveBuffers;
+ this.diskTierExclusiveBuffers = diskTierExclusiveBuffers;
+ this.remoteTierExclusiveBuffers = remoteTierExclusiveBuffers;
+ this.accumulatorExclusiveBuffers = accumulatorExclusiveBuffers;
+ this.memoryTierNumBytesPerSegment = memoryTierNumBytesPerSegment;
+ this.diskTierNumBytesPerSegment = diskTierNumBytesPerSegment;
+ this.remoteTierNumBytesPerSegment = remoteTierNumBytesPerSegment;
+ this.numBuffersTriggerFlushRatio = numBuffersTriggerFlushRatio;
+ this.diskIOSchedulerMaxBuffersReadAhead = diskIOSchedulerMaxBuffersReadAhead;
+ this.diskIOSchedulerRequestTimeout = diskIOSchedulerRequestTimeout;
+ this.minReserveSpaceFraction = minReserveSpaceFraction;
+ this.tierFactories = tierFactories;
}
- public List<TierFactory> getTierFactories() {
- return Arrays.asList(DEFAULT_MEMORY_DISK_TIER_FACTORIES);
+ public static Builder builder(String remoteStorageBasePath) {
+ return new TieredStorageConfiguration.Builder()
+ .setRemoteStorageBasePath(remoteStorageBasePath);
}
- @Nullable
+ public static Builder builder(int tieredStorageBufferSize, String remoteStorageBasePath) {
+ return new TieredStorageConfiguration.Builder()
+ .setTieredStorageBufferSize(tieredStorageBufferSize)
+ .setRemoteStorageBasePath(remoteStorageBasePath);
+ }
+
+ /**
+ * Get the base path on remote storage.
+ *
+ * @return string if the remote storage path is configured otherwise null.
+ */
public String getRemoteStorageBasePath() {
return remoteStorageBasePath;
}
- public static TieredStorageConfiguration.Builder builder() {
- return new TieredStorageConfiguration.Builder();
+ /**
+ * Get the buffer size in tiered storage.
+ *
+ * @return the buffer size.
+ */
+ public int getTieredStorageBufferSize() {
+ return tieredStorageBufferSize;
+ }
+
+ /**
+ * Get exclusive buffer number of memory tier.
+ *
+ * @return the buffer number.
+ */
+ public int getMemoryTierExclusiveBuffers() {
+ return memoryTierExclusiveBuffers;
+ }
+
+ /**
+ * Get exclusive buffer number of disk tier.
+ *
+ * @return the buffer number.
+ */
+ public int getDiskTierExclusiveBuffers() {
+ return diskTierExclusiveBuffers;
+ }
+
+ /**
+ * Get exclusive buffer number of remote tier.
+ *
+ * @return the buffer number.
+ */
+ public int getRemoteTierExclusiveBuffers() {
+ return remoteTierExclusiveBuffers;
}
- /** Builder for {@link TieredStorageConfiguration}. */
+ /**
+ * Get exclusive buffer number of accumulator.
+ *
+ * <p>The buffer number is used to compare with the subpartition number to determine the type of
+ * {@link BufferAccumulator}.
+ *
+ * <p>If the exclusive buffer number is larger than (subpartitionNum + 1), the accumulator will
+ * use {@link HashBufferAccumulator}. If the exclusive buffer number is equal to or smaller than
+ * (subpartitionNum + 1), the accumulator will use {@link SortBufferAccumulator}
+ *
+ * @return the buffer number.
+ */
+ public int getAccumulatorExclusiveBuffers() {
+ return accumulatorExclusiveBuffers;
+ }
+
+ /**
+ * Get the segment size of memory tier.
+ *
+ * @return segment size.
+ */
+ public int getMemoryTierNumBytesPerSegment() {
+ return memoryTierNumBytesPerSegment;
+ }
+
+ /**
+ * Get the segment size of disk tier.
+ *
+ * @return segment size.
+ */
+ public int getDiskTierNumBytesPerSegment() {
+ return diskTierNumBytesPerSegment;
+ }
+
+ /**
+ * Get the segment size of remote tier.
+ *
+ * @return segment size.
+ */
+ public int getRemoteTierNumBytesPerSegment() {
+ return remoteTierNumBytesPerSegment;
+ }
+
+ /**
+ * When the number of buffers that have been requested exceeds this threshold, trigger the
+ * flushing operation in each {@link TierProducerAgent}.
+ *
+ * @return flush ratio.
+ */
+ public float getNumBuffersTriggerFlushRatio() {
+ return numBuffersTriggerFlushRatio;
+ }
+
+ /**
+ * The number of buffers to read ahead at most for each subpartition in {@link DiskIOScheduler},
+ * which can be used to prevent other consumers from starving.
+ *
+ * @return buffer number.
+ */
+ public int getDiskIOSchedulerMaxBuffersReadAhead() {
+ return diskIOSchedulerMaxBuffersReadAhead;
+ }
+
+ /**
+ * Maximum time to wait when requesting read buffers from the buffer pool before throwing an
+ * exception in {@link DiskIOScheduler}.
+ *
+ * @return timeout duration.
+ */
+ public Duration getDiskIOSchedulerBufferRequestTimeout() {
+ return diskIOSchedulerRequestTimeout;
+ }
+
+ /**
+ * Minimum reserved disk space fraction in disk tier.
+ *
+ * @return the fraction.
+ */
+ public float getMinReserveSpaceFraction() {
+ return minReserveSpaceFraction;
+ }
+
+ /**
+ * Get the total exclusive buffer number.
+ *
+ * @return the total exclusive buffer number.
+ */
+ public int getTotalExclusiveBufferNum() {
+ return accumulatorExclusiveBuffers
+ + memoryTierExclusiveBuffers
+ + diskTierExclusiveBuffers
+ + (remoteStorageBasePath == null ? 0 : remoteTierExclusiveBuffers);
+ }
+
+ /**
+ * Get exclusive buffer number of each tier.
+ *
+ * @return buffer number of each tier.
+ */
+ public List<Integer> getEachTierExclusiveBufferNum() {
+ List<Integer> tierExclusiveBuffers = new ArrayList<>();
+ tierExclusiveBuffers.add(memoryTierExclusiveBuffers);
+ tierExclusiveBuffers.add(diskTierExclusiveBuffers);
+ if (remoteStorageBasePath != null) {
+ tierExclusiveBuffers.add(remoteTierExclusiveBuffers);
+ }
+ return tierExclusiveBuffers;
+ }
+
+ public List<TierFactory> getTierFactories() {
+ return tierFactories;
+ }
Review Comment:
These are not consistent. `tierFactories` may or may not be the default factories, while the number of buffers for each tier is hard coded. I think at least we should add a `createDefaultTierExclusiveBufferNums` in the builder, so that we use the default list of buffer numbers only if we use the default factories. A better way might be have a `createDefultTierFactoriesAndMemorySpecs` that decide the hard coded factories and memory parameters at once.
--
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] WencongLiu commented on pull request #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "WencongLiu (via GitHub)" <gi...@apache.org>.
WencongLiu commented on PR #22975:
URL: https://github.com/apache/flink/pull/22975#issuecomment-1641473204
@flinkbot run azure
--
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 #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #22975:
URL: https://github.com/apache/flink/pull/22975#discussion_r1260841837
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -322,7 +322,8 @@ public class NettyShuffleEnvironmentOptions {
.intType()
.defaultValue(1024)
.withDescription(
- "Controls the segment size(in bytes) of hybrid spilled file data index.");
+ "Controls the segment size(in bytes) of hybrid spilled file data index."
+ + "This option will be ignored in new mode of hybrid shuffle.");
Review Comment:
Better points to the corresponding config key.
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -379,6 +381,36 @@ public class NettyShuffleEnvironmentOptions {
+ "tasks have occupied all the buffers and the downstream tasks are waiting for the exclusive buffers. The timeout breaks"
+ "the tie by failing the request of exclusive buffers and ask users to increase the number of total buffers.");
+ /** The option to configure the base remote storage path for the new mode of hybrid shuffle. */
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ public static final ConfigOption<String> NETWORK_HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH =
+ key("taskmanager.network.hybrid-shuffle.remote.path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The option is used to configure the base path of remote storage for the new mode of hybrid shuffle. The new shuffle mode will store"
+ + "the shuffle data in remote storage when the disk space is not enough."
+ + "If the option is configured and the new shuffle mode is disabled, this option will be ignored."
+ + "If the option is not configured and the new shuffle mode is enabled, the remote storage will be disabled in new mode");
+
+ /** The option to enable the new mode of hybrid shuffle. */
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ public static final ConfigOption<Boolean> NETWORK_HYBRID_SHUFFLE_ENABLE_NEW_MODE =
+ ConfigOptions.key("taskmanager.network.hybrid-shuffle.enable-new-mode")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "The option is used to enable the new mode of hybrid shuffle. The following are the differences between new mode and legacy mode:"
+ + "1. Execution efficiency: the new mode has optimized efficiency and will be faster than the legacy mode in many cases."
+ + "2. Job stability: "
+ + "(a) the new mode uses less required network memory. For large parallelism jobs, less network memory is required."
+ + "(b) the new mode can store shuffle data in remote storage when the disk space is not enough, which will avoid insufficient"
+ + "disk space errors. It's supported only when the"
+ + NETWORK_HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH.key()
+ + " is configured.");
Review Comment:
The indention looks good in codes. You may want to check how the document built from this look like.
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -379,6 +381,36 @@ public class NettyShuffleEnvironmentOptions {
+ "tasks have occupied all the buffers and the downstream tasks are waiting for the exclusive buffers. The timeout breaks"
+ "the tie by failing the request of exclusive buffers and ask users to increase the number of total buffers.");
+ /** The option to configure the base remote storage path for the new mode of hybrid shuffle. */
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ public static final ConfigOption<String> NETWORK_HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH =
+ key("taskmanager.network.hybrid-shuffle.remote.path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The option is used to configure the base path of remote storage for the new mode of hybrid shuffle. The new shuffle mode will store"
+ + "the shuffle data in remote storage when the disk space is not enough."
+ + "If the option is configured and the new shuffle mode is disabled, this option will be ignored."
+ + "If the option is not configured and the new shuffle mode is enabled, the remote storage will be disabled in new mode");
Review Comment:
Should points to the corresponding config key. Also, we need to mention the new / legacy mode, but we probably also want to minimize the changes needed later when removing the legacy mode.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java:
##########
@@ -253,6 +348,46 @@ public ResultPartition create(
return partition;
}
+ private List<TieredStorageMemorySpec> createTieredStorageMemorySpecs(
+ int numberOfSubpartitions,
+ BufferAccumulator bufferAccumulator,
+ List<TierProducerAgent> tierProducerAgents) {
+ List<TieredStorageMemorySpec> tieredStorageMemorySpecs = new ArrayList<>();
+ tieredStorageMemorySpecs.add(
+ new TieredStorageMemorySpec(
+ bufferAccumulator,
+ Math.min(
+ numberOfSubpartitions + 1,
+ checkNotNull(tieredStorageConfiguration)
+ .getAccumulatorExclusiveBuffers())));
+ for (TierProducerAgent tierProducerAgent : tierProducerAgents) {
+ if (tierProducerAgent.getClass() == MemoryTierProducerAgent.class) {
+ tieredStorageMemorySpecs.add(
+ new TieredStorageMemorySpec(
+ tierProducerAgent,
+ checkNotNull(tieredStorageConfiguration)
+ .getMemoryTierExclusiveBuffers()));
+ }
+
+ if (tierProducerAgent.getClass() == DiskTierProducerAgent.class) {
+ tieredStorageMemorySpecs.add(
+ new TieredStorageMemorySpec(
+ tierProducerAgent,
+ checkNotNull(tieredStorageConfiguration)
+ .getDiskTierExclusiveBuffers()));
+ }
+
+ if (tierProducerAgent.getClass() == RemoteTierProducerAgent.class) {
+ tieredStorageMemorySpecs.add(
+ new TieredStorageMemorySpec(
+ tierProducerAgent,
+ checkNotNull(tieredStorageConfiguration)
+ .getRemoteTierExclusiveBuffers()));
+ }
Review Comment:
This doesn't seem right. We should not bind the memory spec configuration to the agent classes. Instead, we should create both the factories and memory specs from configuration, possibility in `TieredStorageConfiguration#createDefaultFactories`.
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -379,6 +381,36 @@ public class NettyShuffleEnvironmentOptions {
+ "tasks have occupied all the buffers and the downstream tasks are waiting for the exclusive buffers. The timeout breaks"
+ "the tie by failing the request of exclusive buffers and ask users to increase the number of total buffers.");
+ /** The option to configure the base remote storage path for the new mode of hybrid shuffle. */
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ public static final ConfigOption<String> NETWORK_HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH =
+ key("taskmanager.network.hybrid-shuffle.remote.path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The option is used to configure the base path of remote storage for the new mode of hybrid shuffle. The new shuffle mode will store"
+ + "the shuffle data in remote storage when the disk space is not enough."
+ + "If the option is configured and the new shuffle mode is disabled, this option will be ignored."
+ + "If the option is not configured and the new shuffle mode is enabled, the remote storage will be disabled in new mode");
+
+ /** The option to enable the new mode of hybrid shuffle. */
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ public static final ConfigOption<Boolean> NETWORK_HYBRID_SHUFFLE_ENABLE_NEW_MODE =
+ ConfigOptions.key("taskmanager.network.hybrid-shuffle.enable-new-mode")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "The option is used to enable the new mode of hybrid shuffle. The following are the differences between new mode and legacy mode:"
+ + "1. Execution efficiency: the new mode has optimized efficiency and will be faster than the legacy mode in many cases."
+ + "2. Job stability: "
+ + "(a) the new mode uses less required network memory. For large parallelism jobs, less network memory is required."
+ + "(b) the new mode can store shuffle data in remote storage when the disk space is not enough, which will avoid insufficient"
+ + "disk space errors. It's supported only when the"
+ + NETWORK_HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH.key()
+ + " is configured.");
Review Comment:
We should also mention that this option will be removed in future, as well as the legacy mode.
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -379,6 +381,36 @@ public class NettyShuffleEnvironmentOptions {
+ "tasks have occupied all the buffers and the downstream tasks are waiting for the exclusive buffers. The timeout breaks"
+ "the tie by failing the request of exclusive buffers and ask users to increase the number of total buffers.");
+ /** The option to configure the base remote storage path for the new mode of hybrid shuffle. */
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ public static final ConfigOption<String> NETWORK_HYBRID_SHUFFLE_REMOTE_STORAGE_BASE_PATH =
+ key("taskmanager.network.hybrid-shuffle.remote.path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The option is used to configure the base path of remote storage for the new mode of hybrid shuffle. The new shuffle mode will store"
+ + "the shuffle data in remote storage when the disk space is not enough."
+ + "If the option is configured and the new shuffle mode is disabled, this option will be ignored."
+ + "If the option is not configured and the new shuffle mode is enabled, the remote storage will be disabled in new mode");
+
+ /** The option to enable the new mode of hybrid shuffle. */
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+ @Experimental
+ public static final ConfigOption<Boolean> NETWORK_HYBRID_SHUFFLE_ENABLE_NEW_MODE =
+ ConfigOptions.key("taskmanager.network.hybrid-shuffle.enable-new-mode")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "The option is used to enable the new mode of hybrid shuffle. The following are the differences between new mode and legacy mode:"
+ + "1. Execution efficiency: the new mode has optimized efficiency and will be faster than the legacy mode in many cases."
Review Comment:
Are we sure about this? What cases exactly?
--
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] WencongLiu commented on a diff in pull request #22975: [FLINK-31643][network] Introduce the configurations for tiered storage
Posted by "WencongLiu (via GitHub)" <gi...@apache.org>.
WencongLiu commented on code in PR #22975:
URL: https://github.com/apache/flink/pull/22975#discussion_r1270309221
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java:
##########
@@ -18,58 +18,435 @@
package org.apache.flink.runtime.io.network.partition.hybrid.tiered.common;
-import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.BufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.HashBufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.SortBufferAccumulator;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.disk.DiskIOScheduler;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.disk.DiskTierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.memory.MemoryTierFactory;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.remote.RemoteTierFactory;
-import javax.annotation.Nullable;
-
-import java.util.Arrays;
+import java.time.Duration;
+import java.util.ArrayList;
import java.util.List;
/** Configurations for the Tiered Storage. */
public class TieredStorageConfiguration {
Review Comment:
Fixed.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java:
##########
@@ -164,7 +164,10 @@ private static ResultPartition createResultPartition(
false,
0,
256,
- Long.MAX_VALUE);
+ Long.MAX_VALUE,
Review Comment:
Fixed.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/NettyShuffleUtilsTest.java:
##########
@@ -99,6 +100,8 @@ public void testComputeRequiredNetworkBuffers() throws Exception {
maxRequiredBuffersPerGate,
sortShuffleMinParallelism,
numSortShuffleMinBuffers,
+ false,
+ tieredStoreExclusiveBuffers,
Review Comment:
Fixed.
--
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