[BEAM-4152] Implement session window merging in the Go direct runner #15196
Conversation
|
R: @lostluck |
There was a problem hiding this comment.
Pretty good start! Some style nits, and probably unnecessary duplication to clean up. I've suggested a few rewrites. Consider adding a unit or two for the window merging code in mergeWindows. It's tricky enough by itself to not rest on the rest of the repo to report a problem. Note, it doesn't need to follow the full lifecycle of a CoGBK node, just set up initial conditions properly in the fields and check for the expected results.
| sort.Slice(n.wins, func(i int, j int) bool { | ||
| return n.wins[i].MaxTimestamp() < n.wins[j].MaxTimestamp() | ||
| }) | ||
| n.mergeMap = make(map[typex.Window]int) |
There was a problem hiding this comment.
Likely won't matter for Direct runner purposes, but since mergeWindows and reprocessByWindow are both called in FinishBundle (that is, it doesn't cross a Lifecycle Method boundary), have mergeWindow return the merge map, and reprocessByWindow take it as an argument.
This avoids the CoGBK value holding onto the map with all the window information, allowing it to be garbage collected sooner, lowing memory overhead.
I'm ambivalent about whether the sorted and merged window slice is given the same treatment, as long as it gets cleared away by the end of FinishBundle (similarly for Garbage collecting purposes).
Basically, this is especially important when a pipeline can have arbitrary amounts of GBK operations. Since the Go Direct Runner will always run in a single machine's memory, it pays not to be wasteful and let things get cleaned up.
There was a problem hiding this comment.
It would be worth a comment on mergedMap saying it's a map from the original windows to the index of the new window in the mergedWins slice.
There was a problem hiding this comment.
Done. I figured using some memory to store the map was favorable to slowing things down by needing a binary search implementation to assign the original windows to the merged versions, but tweaking where that is kept to make GC happen sooner is a reasonable improvement.
| for i := 0; i < len(n.wins); { | ||
| intWin, ok := n.wins[i].(window.IntervalWindow) | ||
| if !ok { | ||
| return errors.Errorf("tried to merge non-interval window type") |
There was a problem hiding this comment.
It's unlikely to crop up, but consider printing the actual type of the window in question using the %T formatting directive, and including that in the error message.
See https://play.golang.org/p/8GhLJucF0lA
| newGroups := make(map[string]*group) | ||
| for _, g := range n.m { | ||
| ws := []typex.Window{n.wins[n.mergeMap[g.key.Windows[0]]]} | ||
| var buf bytes.Buffer |
There was a problem hiding this comment.
Consider a helper function or two to reduce the duplication between this and ProcessElement.
Ideally we simply merge the two paths (so delaying work in process element to FinishBundle), but until then, better to clean things up a little via helper functions or helper types.
In this case, for example, the main difference is one is using m.g and this one is using newGroups. A helper type for grouping type grouper map[string]*group could handle this iteration in a method, avoiding the duplication. It just needs the right window information passed in somehow, in both cases... But two or so helper methods would be good, one for the key encoding and one for inserting into the group. Since the method receiver is a map technically, it should use a value rather than a pointer, as maps, even as underlying types are reference types.
There was a problem hiding this comment.
Added helper functions for the key encoding and getting the group from the maps.
|
retest this please |
…pache#15196) * [BEAM-4152] Enable WindowSums tests for the Go direct runner * [BEAM-4152] Implement window merging for the Go direct runner * [BEAM-4152] Refactor mregeWindows to return mergeMap, not keep it in CoGBK struct * [BEAM-4152] Update error message to include window type * [BEAM-4152] Add unit testing for mergeWindows() functionality * [BEAM-4152] Add Apache license to gbk_test.go * [BEAM-4152] Split out key encoding logic into helper for code reuse * [BEAM-4152] Fix formatting * [BEAM-4152] Refactor group map population into helper
…pache#15196) * [BEAM-4152] Enable WindowSums tests for the Go direct runner * [BEAM-4152] Implement window merging for the Go direct runner * [BEAM-4152] Refactor mregeWindows to return mergeMap, not keep it in CoGBK struct * [BEAM-4152] Update error message to include window type * [BEAM-4152] Add unit testing for mergeWindows() functionality * [BEAM-4152] Add Apache license to gbk_test.go * [BEAM-4152] Split out key encoding logic into helper for code reuse * [BEAM-4152] Fix formatting * [BEAM-4152] Refactor group map population into helper
Implements window merging for the session windowing strategy on the Go direct runner. Also re-enables the WindowSums integration test on the direct runner to verify that the implementation works.
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
R: @username).[BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replaceBEAM-XXXwith the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
ValidatesRunnercompliance status (on master branch)Examples testing status on various runners
Post-Commit SDK/Transform Integration Tests Status (on master branch)
Pre-Commit Tests Status (on master branch)
See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI.