[
https://issues.apache.org/jira/browse/FLINK-2897?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15045029#comment-15045029
]
ASF GitHub Bot commented on FLINK-2897:
---------------------------------------
Github user greghogan commented on a diff in the pull request:
https://github.com/apache/flink/pull/1292#discussion_r46828124
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/operators/shipping/OutputEmitter.java
---
@@ -143,16 +148,24 @@ public OutputEmitter(ShipStrategyType strategy,
TypeComparator<T> comparator, Pa
//
--------------------------------------------------------------------------------------------
+ private int[] forward() {
+ return this.channels;
+ }
+
private int[] robin(int numberOfChannels) {
- if (this.channels == null || this.channels.length != 1) {
- this.channels = new int[1];
+ int nextChannel = this.nextChannelToSendTo;
+
+ if (nextChannel >= numberOfChannels) {
+ if (nextChannel == numberOfChannels) {
+ nextChannel = 0;
--- End diff --
The modulo computation is executed in two cases: 1) the initial call to
robin when the source parallelism is greater than the receiver parallelism and
2) when the receiver parallelism is dynamically changing.
Unoptimized modulo is much slower than a test for equality but is required
in the first case to evenly distribute the first outputs.
> Use distinct initial indices for OutputEmitter round-robin
> ----------------------------------------------------------
>
> Key: FLINK-2897
> URL: https://issues.apache.org/jira/browse/FLINK-2897
> Project: Flink
> Issue Type: Improvement
> Components: Distributed Runtime
> Affects Versions: 0.10.0
> Reporter: Greg Hogan
> Assignee: Greg Hogan
>
> Currently, when performing a round-robin partitioning each task will
> sequentially partition starting with partition "1". This is fine in the usual
> case where the number of partitioned objects greatly exceeds the number of
> channels. However, in the case where the number of objects is relatively few
> (each, perhaps, requiring a large computation or access to an external
> system) it would be much better to begin partitioning at distinct indices
> (the task index).
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
