Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1589#discussion_r52012134
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/partitioner/RescalePartitioner.java
---
@@ -0,0 +1,65 @@
+/*
+ * 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.streaming.runtime.partitioner;
+
+import org.apache.flink.runtime.jobgraph.DistributionPattern;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+/**
+ * Partitioner that distributes the data equally by cycling through the
output
+ * channels. This rebalances only to a subset of downstream nodes because
+ * {@link org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator}
instantiates
+ * a {@link DistributionPattern#POINTWISE} distribution pattern when
encountering
+ * {@code SemiRebalancePartitioner}.
+ *
+ * <p>The subset of downstream operations to which the upstream operation
sends
+ * elements depends on the degree of parallelism of both the upstream and
downstream operation.
+ * For example, if the upstream operation has parallelism 2 and the
downstream operation
+ * has parallelism 4, then one upstream operation would distribute
elements to two
+ * downstream operations while the other upstream operation would
distribute to the other
+ * two downstream operations. If, on the other hand, the downstream
operation has parallelism
+ * 2 while the upstream operation has parallelism 4 then two upstream
operations will
+ * distribute to one downstream operation while the other two upstream
operations will
+ * distribute to the other downstream operations.
+ *
+ * <p>In cases where the different parallelisms are not multiples of each
other one or several
+ * downstream operations will have a differing number of inputs from
upstream operations.
+ *
+ * @param <T> Type of the elements in the Stream being rebalanced
+ */
+public class RescalePartitioner<T> extends StreamPartitioner<T> {
+ private static final long serialVersionUID = 1L;
+
+ private int[] returnArray = new int[] {-1};
+
+ @Override
+ public int[] selectChannels(SerializationDelegate<StreamRecord<T>>
record, int numberOfOutputChannels) {
+ this.returnArray[0] = (this.returnArray[0] + 1) %
numberOfOutputChannels;
+ return this.returnArray;
+ }
+
+ public StreamPartitioner<T> copy() {
+ return this;
+ }
+
+ @Override
+ public String toString() {
+ return "SEMI_REBALANCE";
--- End diff --
Shouldn't it be rescale?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
