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:
us...@infra.apache.org


With regards,
Apache Git Services

Reply via email to