You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "SteNicholas (via GitHub)" <gi...@apache.org> on 2024/11/20 15:04:07 UTC
[PR] [CELEBORN-1700] Flink supports fallback to vanilla Flink built-in shuffle implementation [celeborn]
SteNicholas opened a new pull request, #2932:
URL: https://github.com/apache/celeborn/pull/2932
### What changes were proposed in this pull request?
Flink supports fallback to vanilla Flink built-in shuffle implementation.
### Why are the changes needed?
When quota is unenough or workers are unavailable, `RemoteShuffleMaster` does not support fallback to `NettyShuffleMaster`, and `RemoteShuffleEnvironment` does not support fallback to `NettyShuffleEnvironment` at present. Flink should support fallback to vanilla Flink built-in shuffle implementation for unenough quota and unavailable workers.
### Does this PR introduce _any_ user-facing change?
- Introduce `ShuffleFallbackPolicy` interface to determine whether fallback to vanilla Flink built-in shuffle implementation.
```
/**
* The shuffle fallback policy determines whether fallback to vanilla Flink built-in shuffle
* implementation.
*/
public interface ShuffleFallbackPolicy {
/**
* Returns whether fallback to vanilla flink built-in shuffle implementation.
*
* @param shuffleContext The job shuffle context of Flink.
* @param celebornConf The configuration of Celeborn.
* @param lifecycleManager The {@link LifecycleManager} of Celeborn.
* @return Whether fallback to vanilla flink built-in shuffle implementation.
*/
boolean needFallback(
JobShuffleContext shuffleContext,
CelebornConf celebornConf,
LifecycleManager lifecycleManager);
}
```
- Introduce `celeborn.client.flink.shuffle.fallback.policy` config to support shuffle fallback policy configuration.
### How was this patch tested?
`WordCountTestBase#celeborn flink integration test with fallback - word count`
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1700] Flink supports fallback to vanilla Flink built-in shuffle implementation [celeborn]
Posted by "codenohup (via GitHub)" <gi...@apache.org>.
codenohup commented on code in PR #2932:
URL: https://github.com/apache/celeborn/pull/2932#discussion_r1851261674
##########
client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/AbstractRemoteShuffleEnvironment.java:
##########
@@ -129,10 +129,6 @@ public ShuffleIOOwnerContext createShuffleIOOwnerContext(
nettyGroup.addGroup(METRIC_GROUP_INPUT));
}
- public void releasePartitionsLocally(Collection<ResultPartitionID> partitionIds) {
- throw new FlinkRuntimeException("Not implemented yet.");
- }
-
public Collection<ResultPartitionID> getPartitionsOccupyingLocalResources() {
Review Comment:
It may cause the TaskManager to be released unexpectedly, and the shuffle data on the TaskManager will be lost.
The method getPartitionsOccupyingLocalResources of ShuffleEnvironment represents whether the TaskManager should be released when it becomes idle. In case the upstream vertex has finished, and the downstream vertex has not started due to insufficient resources, the TaskManager of the upstream vertex will become idle and be released unexpectedly.
IMO, the method should return partitionManager#getUnreleasedPartitions, as the ResultPartition created by the Flink framework (such as SortMergeResultPartition) will be registered to the partitionManager, and the ResultPartition created by Celeborn (RemoteShuffleResultPartition) is not registered with the partitionManager, as it does not occupy TaskManager resources.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1700] Flink supports fallback to vanilla Flink built-in shuffle implementation [celeborn]
Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong commented on code in PR #2932:
URL: https://github.com/apache/celeborn/pull/2932#discussion_r1860382139
##########
tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/WordCountTest.scala:
##########
@@ -65,13 +66,22 @@ abstract class WordCountTestBase extends AnyFunSuite with Logging with MiniClust
}
test(getClass.getName + ": celeborn flink integration test - word count") {
+ testWordCount(FallbackPolicy.AUTO)
+ }
+
+ test(getClass.getName + ": celeborn flink integration test with fallback - word count") {
+ testWordCount(FallbackPolicy.ALWAYS)
Review Comment:
Could we verify if Flink uses its built-in shuffle when the fallback.policy is set to FallbackPolicy.ALWAYS?
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1700] Flink supports fallback to vanilla Flink built-in shuffle implementation [celeborn]
Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong closed pull request #2932: [CELEBORN-1700] Flink supports fallback to vanilla Flink built-in shuffle implementation
URL: https://github.com/apache/celeborn/pull/2932
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1700] Flink supports fallback to vanilla Flink built-in shuffle implementation [celeborn]
Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong commented on PR #2932:
URL: https://github.com/apache/celeborn/pull/2932#issuecomment-2503916153
Thanks, merge to main(v0.6.0)
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1700] Flink supports fallback to vanilla Flink built-in shuffle implementation [celeborn]
Posted by "codenohup (via GitHub)" <gi...@apache.org>.
codenohup commented on code in PR #2932:
URL: https://github.com/apache/celeborn/pull/2932#discussion_r1851261674
##########
client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/AbstractRemoteShuffleEnvironment.java:
##########
@@ -129,10 +129,6 @@ public ShuffleIOOwnerContext createShuffleIOOwnerContext(
nettyGroup.addGroup(METRIC_GROUP_INPUT));
}
- public void releasePartitionsLocally(Collection<ResultPartitionID> partitionIds) {
- throw new FlinkRuntimeException("Not implemented yet.");
- }
-
public Collection<ResultPartitionID> getPartitionsOccupyingLocalResources() {
Review Comment:
The `getPartitionsOccupyingLocalResources` return empty list may cause the TaskManager to be released unexpectedly, and the shuffle data on the TaskManager will be lost.
The method `getPartitionsOccupyingLocalResources` of `ShuffleEnvironment` represents whether the TaskManager should be released when it becomes idle. In case the upstream vertex has finished, and the downstream vertex has not started due to insufficient resources, the TaskManager of the upstream vertex will become idle and be released unexpectedly.
IMO, the method should return `partitionManager#getUnreleasedPartitions`, as the `ResultPartition` created by the Flink framework (such as `SortMergeResultPartition`) will be registered to the `partitionManager`, and the `ResultPartition` created by Celeborn (`RemoteShuffleResultPartition`) is not registered with the `partitionManager`, as it does not occupy TaskManager resources.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1700] Flink supports fallback to vanilla Flink built-in shuffle implementation [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #2932:
URL: https://github.com/apache/celeborn/pull/2932#issuecomment-2489123388
Ping @reswqa, @RexXiong.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1700] Flink supports fallback to vanilla Flink built-in shuffle implementation [celeborn]
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #2932:
URL: https://github.com/apache/celeborn/pull/2932#discussion_r1851510694
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -1040,8 +1040,13 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
def shuffleRangeReadFilterEnabled: Boolean = get(SHUFFLE_RANGE_READ_FILTER_ENABLED)
def shuffleForceFallbackEnabled: Boolean = get(SPARK_SHUFFLE_FORCE_FALLBACK_ENABLED)
def checkWorkerEnabled: Boolean = get(CHECK_WORKER_ENABLED)
- def shuffleFallbackPolicy: FallbackPolicy = {
- val fallbackPolicyGiven = FallbackPolicy.valueOf(get(SPARK_SHUFFLE_FALLBACK_POLICY))
+ def sparkShuffleFallbackPolicy: FallbackPolicy = {
+ shuffleFallbackPolicy(FallbackPolicy.valueOf(get(SPARK_SHUFFLE_FALLBACK_POLICY)))
+ }
+ def flinkShuffleFallbackPolicy: FallbackPolicy = {
+ shuffleFallbackPolicy(FallbackPolicy.valueOf(get(Flink_SHUFFLE_FALLBACK_POLICY)))
Review Comment:
```suggestion
shuffleFallbackPolicy(FallbackPolicy.valueOf(get(FLINK_SHUFFLE_FALLBACK_POLICY)))
```
##########
client-flink/flink-1.20/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleServiceFactory.java:
##########
@@ -17,22 +17,27 @@
package org.apache.celeborn.plugin.flink;
+import org.apache.flink.runtime.io.network.NettyShuffleServiceFactory;
import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.io.network.partition.consumer.IndexedInputGate;
+import org.apache.flink.runtime.shuffle.ShuffleDescriptor;
import org.apache.flink.runtime.shuffle.ShuffleEnvironment;
import org.apache.flink.runtime.shuffle.ShuffleEnvironmentContext;
import org.apache.flink.runtime.shuffle.ShuffleMaster;
import org.apache.flink.runtime.shuffle.ShuffleMasterContext;
import org.apache.flink.runtime.shuffle.ShuffleServiceFactory;
public class RemoteShuffleServiceFactory extends AbstractRemoteShuffleServiceFactory
- implements ShuffleServiceFactory<
- RemoteShuffleDescriptor, ResultPartitionWriter, IndexedInputGate> {
+ implements ShuffleServiceFactory<ShuffleDescriptor, ResultPartitionWriter, IndexedInputGate> {
Review Comment:
Could we also support fallback for flink-1.20?
##########
client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/fallback/WorkersAvailableFallbackPolicy.java:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.celeborn.plugin.flink.fallback;
+
+import org.apache.flink.runtime.shuffle.JobShuffleContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.client.LifecycleManager;
+import org.apache.celeborn.common.CelebornConf;
+
+public class WorkersAvailableFallbackPolicy implements ShuffleFallbackPolicy {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WorkersAvailableFallbackPolicy.class);
+
+ public static final WorkersAvailableFallbackPolicy INSTANCE =
+ new WorkersAvailableFallbackPolicy();
+
+ /**
+ * If celeborn cluster has no available workers, fallback to flink built-in shuffle
+ * implementation.
+ *
+ * @param shuffleContext The job shuffle context of Flink.
+ * @param celebornConf The configuration of Celeborn.
+ * @param lifecycleManager The {@link LifecycleManager} of Celeborn.
+ * @return Whether celeborn cluster has no available workers.
+ */
+ @Override
+ public boolean needFallback(
+ JobShuffleContext shuffleContext,
+ CelebornConf celebornConf,
+ LifecycleManager lifecycleManager) {
+ if (!celebornConf.checkWorkerEnabled()) {
Review Comment:
We should also update doc for `CHECK_WORKER_ENABLED`.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1700] Flink supports fallback to vanilla Flink built-in shuffle implementation [celeborn]
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #2932:
URL: https://github.com/apache/celeborn/pull/2932#discussion_r1851528942
##########
client-flink/flink-1.20/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleServiceFactory.java:
##########
@@ -17,22 +17,27 @@
package org.apache.celeborn.plugin.flink;
+import org.apache.flink.runtime.io.network.NettyShuffleServiceFactory;
import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.io.network.partition.consumer.IndexedInputGate;
+import org.apache.flink.runtime.shuffle.ShuffleDescriptor;
import org.apache.flink.runtime.shuffle.ShuffleEnvironment;
import org.apache.flink.runtime.shuffle.ShuffleEnvironmentContext;
import org.apache.flink.runtime.shuffle.ShuffleMaster;
import org.apache.flink.runtime.shuffle.ShuffleMasterContext;
import org.apache.flink.runtime.shuffle.ShuffleServiceFactory;
public class RemoteShuffleServiceFactory extends AbstractRemoteShuffleServiceFactory
- implements ShuffleServiceFactory<
- RemoteShuffleDescriptor, ResultPartitionWriter, IndexedInputGate> {
+ implements ShuffleServiceFactory<ShuffleDescriptor, ResultPartitionWriter, IndexedInputGate> {
Review Comment:
Could we also support fallback for flink-1.20?
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1700] Flink supports fallback to vanilla Flink built-in shuffle implementation [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #2932:
URL: https://github.com/apache/celeborn/pull/2932#issuecomment-2497073196
Ping @RexXiong, @FMX.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org