echauchot 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_r397716574
##########
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 sure, I agree about hot keys. But if spark spills to disk it is because
it is out of memory. And this PR does not claim to fix the OOM issue that also
existed in the previous version of GBK but it claims to remove a step in the
translation and avoid memory consumption. I could profile the load test to give
you numbers on memory because it is not integrated in load tests
Also see my comment
https://github.com/apache/beam/pull/11055#discussion_r397246846 regarding
materialization.
Anyway there is hotkey configuration in GBKLoadTest. I'll configure it and
post the results
----------------------------------------------------------------
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
