[
https://issues.apache.org/jira/browse/BEAM-9436?focusedWorklogId=409406&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-409406
]
ASF GitHub Bot logged work on BEAM-9436:
----------------------------------------
Author: ASF GitHub Bot
Created on: 25/Mar/20 09:34
Start Date: 25/Mar/20 09:34
Worklog Time Spent: 10m
Work Description: echauchot commented on 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]
Issue Time Tracking
-------------------
Worklog Id: (was: 409406)
Time Spent: 12.5h (was: 12h 20m)
> Try to avoid elements list materialization in GBK
> -------------------------------------------------
>
> Key: BEAM-9436
> URL: https://issues.apache.org/jira/browse/BEAM-9436
> Project: Beam
> Issue Type: Improvement
> Components: runner-spark
> Reporter: Etienne Chauchot
> Assignee: Etienne Chauchot
> Priority: Major
> Labels: structured-streaming
> Time Spent: 12.5h
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
