zhuzhurk commented on a change in pull request #17952: URL: https://github.com/apache/flink/pull/17952#discussion_r760017845
########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/VertexParallelismDecider.java ########## @@ -0,0 +1,45 @@ +/* + * 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.runtime.scheduler.adaptivebatch; + +import java.util.List; + +/** + * {@link VertexParallelismDecider} is responsible for determining the parallelism of a job vertex, + * based on the size of the consumed blocking results. + */ +public interface VertexParallelismDecider { + + /** + * Computing the parallelism. + * + * @param consumedResults The size of consumed blocking results. Review comment: the param description looks not very accurate ########## File path: flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java ########## @@ -480,6 +480,47 @@ .withDescription( "Controls whether partitions should already be released during the job execution."); + @Documentation.Section({ + Documentation.Sections.EXPERT_SCHEDULING, + Documentation.Sections.ALL_JOB_MANAGER + }) + public static final ConfigOption<Integer> ADAPTIVE_BACH_SCHEDULER_MIN_PARALLELISM = + key("jobmanager.scheduler.adaptive-batch.min-parallelism") + .intType() + .defaultValue(1) + .withDescription("The lower bound of allowed parallelism to set adaptively"); + + @Documentation.Section({ + Documentation.Sections.EXPERT_SCHEDULING, + Documentation.Sections.ALL_JOB_MANAGER + }) + public static final ConfigOption<Integer> ADAPTIVE_BATCH_SCHEDULER_MAX_PARALLELISM = + key("jobmanager.scheduler.adaptive-batch.max-parallelism") + .intType() + .defaultValue(128) + .withDescription("The upper bound of allowed parallelism to set adaptively"); + + @Documentation.Section({ + Documentation.Sections.EXPERT_SCHEDULING, + Documentation.Sections.ALL_JOB_MANAGER + }) + public static final ConfigOption<MemorySize> ADAPTIVE_BATCH_SCHEDULER_DATA_VOLUME_PER_TASK = + key("jobmanager.scheduler.adaptive-batch.data-volume-per-task") + .memoryType() + .defaultValue(MemorySize.ofMebiBytes(1024)) + .withDescription( + "The size of data volume to expect each task instance to process."); + + @Documentation.Section({ + Documentation.Sections.EXPERT_SCHEDULING, + Documentation.Sections.ALL_JOB_MANAGER + }) + public static final ConfigOption<Integer> ADAPTIVE_BATCH_SCHEDULER_DEFAULT_SOURCE_PARALLELISM = + key("jobmanager.scheduler.adaptive-batch.source-parallelism.default") + .intType() + .defaultValue(1) + .withDescription("The default parallelism of source vertices."); Review comment: I think we should add a limitation of the scope of these configs. e.g. The default parallelism of source vertices if %s has been set to %s", SCHEDULER.key(), AdaptiveBatch.name() Similar as what we've done in `RestartStrategyOptions`. ########## File path: flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java ########## @@ -480,6 +480,47 @@ .withDescription( "Controls whether partitions should already be released during the job execution."); + @Documentation.Section({ + Documentation.Sections.EXPERT_SCHEDULING, + Documentation.Sections.ALL_JOB_MANAGER + }) + public static final ConfigOption<Integer> ADAPTIVE_BACH_SCHEDULER_MIN_PARALLELISM = Review comment: BACH -> BATCH ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/BlockingResultInfo.java ########## @@ -0,0 +1,42 @@ +/* + * 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.runtime.scheduler.adaptivebatch; + +import java.util.List; + +/** The blocking result info, which will be used to calculate the vertex parallelism. */ +public class BlockingResultInfo { + + private List<Long> blockingPartitionSizes; Review comment: can be final ########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/DefaultVertexParallelismDeciderTest.java ########## @@ -0,0 +1,117 @@ +/* + * 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.runtime.scheduler.adaptive; + +import org.apache.flink.configuration.MemorySize; +import org.apache.flink.runtime.scheduler.adaptivebatch.BlockingResultInfo; +import org.apache.flink.runtime.scheduler.adaptivebatch.DefaultVertexParallelismDecider; +import org.apache.flink.runtime.scheduler.adaptivebatch.VertexParallelismDecider; + +import org.junit.Test; + +import java.util.Arrays; +import java.util.Collections; + +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.MatcherAssert.assertThat; + +/** Test for {@link DefaultVertexParallelismDecider}. */ +public class DefaultVertexParallelismDeciderTest { + + private static final long BYTE_256_MB = 256 * 1024 * 1024L; + private static final long BYTE_512_MB = 512 * 1024 * 1024L; + private static final long BYTE_1_GB = 1024 * 1024 * 1024L; + private static final long BYTE_2_GB = 2 * 1024 * 1024 * 1024L; + private static final long BYTE_8_GB = 8 * 1024 * 1024 * 1024L; + + private static final int MAX_PARALLELISM = 100; + private static final int MIN_PARALLELISM = 2; + private static final int DEFAULT_SOURCE_PARALLELISM = 10; + + VertexParallelismDecider decider = Review comment: can be private final ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismDecider.java ########## @@ -0,0 +1,132 @@ +/* + * 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.runtime.scheduler.adaptivebatch; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.JobManagerOptions; +import org.apache.flink.configuration.MemorySize; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; + +/** Default implementation of {@link VertexParallelismDecider}. */ +public class DefaultVertexParallelismDecider implements VertexParallelismDecider { + + private static final Logger LOG = + LoggerFactory.getLogger(DefaultVertexParallelismDecider.class); + + private final int maxParallelism; + private final int minParallelism; + private final long dataVolumePerTask; + private final int defaultSourceParallelism; + + public DefaultVertexParallelismDecider( + int maxParallelism, + int minParallelism, + MemorySize dataVolumePerTask, + int defaultSourceParallelism) { + + this.maxParallelism = maxParallelism; + this.minParallelism = minParallelism; + this.dataVolumePerTask = dataVolumePerTask.getBytes(); + this.defaultSourceParallelism = defaultSourceParallelism; + } + + @Override + public int decideParallelismForVertex(List<BlockingResultInfo> consumedResults) { + + if (consumedResults.isEmpty()) { + // source job vertex + return defaultSourceParallelism; + } else { + return calculateParallelism(consumedResults); + } + } + + private int calculateParallelism(List<BlockingResultInfo> consumedResults) { + + long broadcastBytes = + consumedResults.stream() + .filter(BlockingResultInfo::isBroadcast) + .mapToLong( + consumedResult -> + consumedResult.getBlockingPartitionSizes().stream() + .reduce(0L, Long::sum)) + .sum(); + + long nonBroadcastBytes = + consumedResults.stream() + .filter(consumedResult -> !consumedResult.isBroadcast()) + .mapToLong( + consumedResult -> + consumedResult.getBlockingPartitionSizes().stream() + .reduce(0L, Long::sum)) + .sum(); + + if (broadcastBytes > dataVolumePerTask + || (broadcastBytes == dataVolumePerTask && nonBroadcastBytes > 0)) { + LOG.warn( + "The minimum size of one task to process is larger than " + + "the size of data volume which is configured by " + + "'" + + JobManagerOptions.ADAPTIVE_BATCH_SCHEDULER_DATA_VOLUME_PER_TASK.key() + + "'. " + + "Parallelism will be set to {}.", + maxParallelism); + + return maxParallelism; Review comment: if `nonBroadcastBytes` is 0, I think there is no need to set parallelism to a large value. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismDecider.java ########## @@ -0,0 +1,132 @@ +/* + * 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.runtime.scheduler.adaptivebatch; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.JobManagerOptions; +import org.apache.flink.configuration.MemorySize; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; + +/** Default implementation of {@link VertexParallelismDecider}. */ +public class DefaultVertexParallelismDecider implements VertexParallelismDecider { + + private static final Logger LOG = + LoggerFactory.getLogger(DefaultVertexParallelismDecider.class); + + private final int maxParallelism; + private final int minParallelism; + private final long dataVolumePerTask; + private final int defaultSourceParallelism; + + public DefaultVertexParallelismDecider( + int maxParallelism, + int minParallelism, + MemorySize dataVolumePerTask, + int defaultSourceParallelism) { + + this.maxParallelism = maxParallelism; + this.minParallelism = minParallelism; + this.dataVolumePerTask = dataVolumePerTask.getBytes(); + this.defaultSourceParallelism = defaultSourceParallelism; + } + + @Override + public int decideParallelismForVertex(List<BlockingResultInfo> consumedResults) { + + if (consumedResults.isEmpty()) { + // source job vertex + return defaultSourceParallelism; + } else { + return calculateParallelism(consumedResults); + } + } + + private int calculateParallelism(List<BlockingResultInfo> consumedResults) { + + long broadcastBytes = + consumedResults.stream() + .filter(BlockingResultInfo::isBroadcast) + .mapToLong( + consumedResult -> + consumedResult.getBlockingPartitionSizes().stream() + .reduce(0L, Long::sum)) + .sum(); + + long nonBroadcastBytes = + consumedResults.stream() + .filter(consumedResult -> !consumedResult.isBroadcast()) + .mapToLong( + consumedResult -> + consumedResult.getBlockingPartitionSizes().stream() + .reduce(0L, Long::sum)) + .sum(); + + if (broadcastBytes > dataVolumePerTask + || (broadcastBytes == dataVolumePerTask && nonBroadcastBytes > 0)) { + LOG.warn( Review comment: I prefer the log to be INFO because this case is acceptable. Just similar to the case that when calculated parallelism is larger than maxParallelism and limited to maxParallelism, the data processed by each task will also exceed `dataVolumePerTask`. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/VertexParallelismDeciderFactoryLoader.java ########## @@ -0,0 +1,34 @@ +/* + * 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.runtime.scheduler.adaptivebatch; + +import org.apache.flink.configuration.Configuration; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** A utility class to load job vertex parallelism decider factories from the configuration. */ +public class VertexParallelismDeciderFactoryLoader { + + public static VertexParallelismDecider.Factory loadJobVertexParallelismDeciderFactory( + final Configuration configuration) { + checkNotNull(configuration); + // currently, only support one decider. Review comment: and then we do not need to introduce `JobVertexParallelismDecider.Factory` and a factory method `DefaultJobVertexParallelismDecider.from(Configuration)` would be enough. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/VertexParallelismDeciderFactoryLoader.java ########## @@ -0,0 +1,34 @@ +/* + * 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.runtime.scheduler.adaptivebatch; + +import org.apache.flink.configuration.Configuration; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** A utility class to load job vertex parallelism decider factories from the configuration. */ +public class VertexParallelismDeciderFactoryLoader { Review comment: better to create a private constructor for it to avoid it been instantiated. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismDecider.java ########## @@ -0,0 +1,132 @@ +/* + * 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.runtime.scheduler.adaptivebatch; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.JobManagerOptions; +import org.apache.flink.configuration.MemorySize; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; + +/** Default implementation of {@link VertexParallelismDecider}. */ +public class DefaultVertexParallelismDecider implements VertexParallelismDecider { + + private static final Logger LOG = + LoggerFactory.getLogger(DefaultVertexParallelismDecider.class); + + private final int maxParallelism; + private final int minParallelism; + private final long dataVolumePerTask; + private final int defaultSourceParallelism; + + public DefaultVertexParallelismDecider( Review comment: can be package private (or even private by re-work the test) ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/VertexParallelismDeciderFactoryLoader.java ########## @@ -0,0 +1,34 @@ +/* + * 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.runtime.scheduler.adaptivebatch; + +import org.apache.flink.configuration.Configuration; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** A utility class to load job vertex parallelism decider factories from the configuration. */ +public class VertexParallelismDeciderFactoryLoader { + + public static VertexParallelismDecider.Factory loadJobVertexParallelismDeciderFactory( + final Configuration configuration) { + checkNotNull(configuration); + // currently, only support one decider. Review comment: Looks to me there is no need to introduce such a loader at the moment because there is only have `DefaultVertexParallelismDecider`. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
