[BEAM-13015] Update the SDK harness grouping table to be memory bounded based upon the amount of assigned cache memory and to use an LRU eviction policy. #17327

lukecwik · 2022-04-09T05:23:34Z

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

…n amount of assigned cache memory and also use an LRU policy for evicting entries from the table.

lukecwik · 2022-04-09T05:23:43Z

R: @youngoli

lukecwik · 2022-04-12T23:31:49Z

Run Java PreCommit

lukecwik · 2022-04-12T23:31:55Z

Run Python_PVR_Flink PreCommit

lukecwik · 2022-04-13T17:38:19Z

Run Java PreCommit

lukecwik · 2022-04-28T16:27:19Z

R: @Abacn

aaltay · 2022-05-12T21:42:54Z

@Abacn - could you please review this change?

y1chi · 2022-05-13T03:35:58Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/PrecombineGroupingTable.java

+              }
+              return tableEntry;
+            });
+    weight += entry.getWeight();


is this accurate if entry is not new?

Fixed and updated tests since it turned out we weren't accounting for the grouping table key.

y1chi · 2022-05-13T03:42:42Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/PrecombineGroupingTable.java

 @SuppressWarnings({
  "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
 })
+@NotThreadSafe


Document why? Also seems to contradict the requirement of Shrinkable?

Documented that put and flush must be called from the bundle processing thread. shrink can be called from any thread.

y1chi · 2022-05-13T04:09:51Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/PrecombineGroupingTable.java

+
+    // Get the updated weight now that the cache may have been shrunk and respect it
+    long currentMax = maxWeight.get();
+    if (weight > currentMax) {


If this is triggered by shrink() why not do it in shrink but instead rely on new input?

Because we want to make sure that we only produce output from the bundle processing thread and not from an arbitrary thread that caused the shrinking to happen. Added a comment to reflect.

y1chi · 2022-05-13T04:51:22Z

sdks/java/harness/src/test/java/org/apache/beam/fn/harness/PrecombineGroupingTableTest.java


-    table.put("DDDD", 6, receiver);
-    assertThat(receiver.outputElems, hasItem((Object) KV.of("DDDD", 6L)));
+    // Insert three values which even with compaction isn't enough so we evict D & E to get


s/D & E/A & B/

lukecwik · 2022-05-13T22:25:23Z

@y1chi PTAL

y1chi · 2022-05-13T22:54:28Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/PrecombineGroupingTable.java

+        groupingKey,
+        (key, tableEntry) -> {
+          if (tableEntry == null) {
+            weight += groupingKey.getWeight();


remove this?

this adds the weight of the key, and not the value

isn't entry.getWeight() = key.getWeight() + accumulator.getWeight()?

There are two cases.

key == structural key, then:

GroupingTableKey weight = key weight + windows weight + pane info weight

GroupingTableEntry weight = reference weight + accumulator weight

key != structural key, then:

GroupingTableKey weight = structural key weight + windows weight + pane info weight

GroupingTableEntry weight = key weight + accumulator weight

y1chi

LGTM

youngoli

As far as I can tell it looks good, although I had some trouble following all the various weights involved so I'm glad Yichi's here to provide a second set of eyes.

youngoli · 2022-05-14T02:00:45Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/PrecombineGroupingTable.java

+        Iterator<GroupingTableEntry> iterator = lruMap.values().iterator();
+        while (iterator.hasNext()) {
+          GroupingTableEntry valueToFlush = iterator.next();
+          weight -= valueToFlush.getWeight() + valueToFlush.getGroupingKey().getWeight();


I'm having some trouble following all the different weights, and my first instinct is that since valueToFlush contains the GroupingKey, that this would count the weight of the grouping key twice (and presumably this would be bad because it wasn't counted twice when being originally added to the max weight).

lukecwik · 2022-05-15T21:07:58Z

Run Java PreCommit

robertwb · 2022-05-25T01:00:30Z

I happened to do some benchmarking for a separate change (#17641) and noticed that this PR seems to reduce the performance significantly. Before (https://github.com/robertwb/incubator-beam/tree/java-combine-key-old) I was getting stats

  33,102 ±(99.9%) 1,173 ops/s [Average]
  (min, avg, max) = (32,761, 33,102, 33,492), stdev = 0,305
  CI (99.9%): [31,929, 34,275] (assumes normal distribution)

  24,809 ±(99.9%) 0,861 ops/s [Average]
  (min, avg, max) = (24,521, 24,809, 25,083), stdev = 0,224
  CI (99.9%): [23,948, 25,670] (assumes normal distribution)

(two benchmarks here: globally windowed and not) but after merging this change I'm seeing

Result "org.apache.beam.fn.harness.jmh.CombinerTableBenchmark.uniformDistribution":
  4,949 ±(99.9%) 0,349 ops/s [Average]
  (min, avg, max) = (4,832, 4,949, 5,059), stdev = 0,091
  CI (99.9%): [4,601, 5,298] (assumes normal distribution)

Result "org.apache.beam.fn.harness.jmh.CombinerTableBenchmark.uniformDistribution":
  3,855 ±(99.9%) 0,304 ops/s [Average]
  (min, avg, max) = (3,735, 3,855, 3,930), stdev = 0,079
  CI (99.9%): [3,551, 4,159] (assumes normal distribution)

robertwb · 2022-05-25T21:27:56Z

I should note that before either change I was getting on the order of 15k ops/sec.

[BEAM-13015] Update the grouping table to be memory bounded based upo…

a454a6d

…n amount of assigned cache memory and also use an LRU policy for evicting entries from the table.

github-actions bot added core java runners labels Apr 9, 2022

fixup! checkstyle

b871fb6

lukecwik mentioned this pull request May 12, 2022

[BEAM-14464] More efficient grouping keys in precombiner table. #17641

Merged

4 tasks

y1chi reviewed May 13, 2022

View reviewed changes

fixup! Address PR comments.

975a97c

y1chi reviewed May 13, 2022

View reviewed changes

y1chi approved these changes May 13, 2022

View reviewed changes

youngoli approved these changes May 14, 2022

View reviewed changes

lukecwik merged commit 5b81d14 into apache:master May 16, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-13015] Update the SDK harness grouping table to be memory bounded based upon the amount of assigned cache memory and to use an LRU eviction policy. #17327

[BEAM-13015] Update the SDK harness grouping table to be memory bounded based upon the amount of assigned cache memory and to use an LRU eviction policy. #17327

lukecwik commented Apr 9, 2022

lukecwik commented Apr 9, 2022

lukecwik commented Apr 12, 2022

lukecwik commented Apr 12, 2022

lukecwik commented Apr 13, 2022

lukecwik commented Apr 28, 2022 •

edited

Loading

aaltay commented May 12, 2022

y1chi May 13, 2022

lukecwik May 13, 2022

y1chi May 13, 2022

lukecwik May 13, 2022

y1chi May 13, 2022

lukecwik May 13, 2022

y1chi May 13, 2022

lukecwik May 13, 2022

lukecwik commented May 13, 2022

y1chi May 13, 2022

lukecwik May 13, 2022

y1chi May 13, 2022

lukecwik May 13, 2022

y1chi left a comment

youngoli left a comment

youngoli May 14, 2022

lukecwik commented May 15, 2022

robertwb commented May 25, 2022

robertwb commented May 25, 2022

[BEAM-13015] Update the SDK harness grouping table to be memory bounded based upon the amount of assigned cache memory and to use an LRU eviction policy. #17327

[BEAM-13015] Update the SDK harness grouping table to be memory bounded based upon the amount of assigned cache memory and to use an LRU eviction policy. #17327

Conversation

lukecwik commented Apr 9, 2022

GitHub Actions Tests Status (on master branch)

lukecwik commented Apr 9, 2022

lukecwik commented Apr 12, 2022

lukecwik commented Apr 12, 2022

lukecwik commented Apr 13, 2022

lukecwik commented Apr 28, 2022 • edited Loading

aaltay commented May 12, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lukecwik commented May 13, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

y1chi left a comment

Choose a reason for hiding this comment

youngoli left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lukecwik commented May 15, 2022

robertwb commented May 25, 2022

robertwb commented May 25, 2022

lukecwik commented Apr 28, 2022 •

edited

Loading