apache/incubator-flink — pull request #84 by StephanEwen (2014-07-29 15:51:58+00:00)
Title: [FLINK-1023] Switch group-at-a-time function to Iterable (from Iterator)

This patch allows the *GroupReduce* and the *CoGroup* to use the beautiful *foreach* loop syntax.

```java
public void reduce(Iterable values, Collector out)
long sum = 0L;

for (Long num : values) {
sum += num;
}
}
```

Since the data behind the iterable is transient, you cannot iterate over it multiple times. The next time you request an iterator, it will pick up where the previous left off, potentially returning an empty iterator:

```java
public void reduce(Iterable values, Collector out)
for (Long num : values) {
// do something
}

for (Long num : values) {
// empty loop, will never be entered
}
}
```

Some functions need iterator behavior, which can still be used
```java
public void coGroup(Iterable values1, Iterable values2, Collector out)
if (values2.iterator().hasNext()) {
// do something
}
}

public void coGroup(Iterable values1, Iterable values2, Collector out)
Iterator iter = values2.iterator();
if (values2.iterator().hasNext()) {
do {
// something
} while (iter.hasNext());
}
}
```

----

Note:

The *Iterable* and the *Iterator* are currently strictly in sync. That means one can pick up where the other left and vice versa. While I would not encourage it to use this pattern, you can actually mix then and do something like the following (which is equivalent to the first example):
```java
public void reduce(Iterable values, Collector out)
Long sum = values.iterator().next();

for (Long num : values) {
sum += num;
}
}
```