iemejia commented on a change in pull request #11055: [BEAM-9436] Improve GBK
in spark structured streaming runner
URL: https://github.com/apache/beam/pull/11055#discussion_r396491467
##########
File path:
runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/functions/GroupAlsoByWindowViaOutputBufferFn.java
##########
@@ -65,9 +65,15 @@ public GroupAlsoByWindowViaOutputBufferFn(
@Override
public Iterator<WindowedValue<KV<K, Iterable<InputT>>>> call(
- KV<K, Iterable<WindowedValue<InputT>>> kv) throws Exception {
- K key = kv.getKey();
- Iterable<WindowedValue<InputT>> values = kv.getValue();
+ K key, Iterator<WindowedValue<KV<K, InputT>>> iterator) throws Exception
{
+
+ // we have to meterialize the Iterator because
ReduceFnRunner.processElements expects
+ // ArrayList<WindowedValue<InputT>> and not Iterator<WindowedValue<KV<K,
InputT>>>
+ ArrayList<WindowedValue<InputT>> values = new ArrayList<>();
+ while (iterator.hasNext()) {
+ WindowedValue<KV<K, InputT>> wv = iterator.next();
+ values.add(wv.withValue(wv.getValue().getValue()));
Review comment:
> Yes this comment in the doc does seem to confirm this `...users must take
care to avoid materializing the whole iterator for a group (for example, by
calling toList) unless they are sure that this is possible given the memory
constraints of their cluster.` this looks like doing exactly the same than I
would expect the `toList` call to do.
See this comment
https://github.com/apache/beam/pull/11055#discussion_r397246846
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
