[BEAM-2926] Migrate to using a trivial multimap materialization within the Java SDK. by lukecwik · Pull Request #4011 · apache/beam

lukecwik · 2017-10-18T21:40:57Z

Follow this checklist to help us incorporate your contribution quickly and easily:

Make sure there is a JIRA issue filed for the change (usually before you start working on it). Trivial changes like typos do not require a JIRA issue. Your pull request should address just this issue, without pulling in other changes.
Each commit in the pull request should have a meaningful subject line and body.
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue.
Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
Run mvn clean verify to make sure basic checks pass. A more thorough check will be performed on your pull request automatically.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

I want to get this reviewed before I start mucking with the Dataflow worker so I can update that internally after this approach has been agreed upon.

Review notes:

I dropped the "helpers" in PCollectionViewTesting that weren't very useful.
I also migrated off of tests calling PCollectionViews directly and instead constructing them through pc.apply(View.asYYY()) to simplify maintenance since PCollectionViews is internal only view View.asYYY is public.
I tried to migrate uses of PCollectionView.getCoderInternal()/PCollection.getWindowingStrategyInternal()/... to PCollectionView.getPCollection().getYYY() but was unable to because the PCollection stored within PCollectionView is transient and the majority of runners pull information from serialized PCollectionViews, not from proto round tripped views.
I loosened the types on PCollectionView so that they weren't so "Iterable" materialization focused.
I thought of creating a MultimapPCollectionView or something like that which narrowed the types of PCollectionView so it was easier to use PCollectionView instead of all the ? types and forced casts but that would have grown this PR to be so much larger.

lukecwik · 2017-10-19T16:46:54Z

R: @tgroh
CC: @kennknowles

I had a brief discussion with Ken about doing this high level approach since he had been the largest contributor to the PCollectionView interfaces from my memory. Ken, you might want to take a look at the high level interface changes.

tgroh · 2017-10-19T19:46:12Z

sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PCollectionViewTesting.java

+  public static List<Object> materializeValuesFor(
+      PTransform<?, ? extends PCollectionView<?>> viewTransformClass, Object ... values) {
+    List<Object> rval = new ArrayList<>();
+    // Currently all view materializations are the same where the data is shared underneath


Can this be any more specific than a PTransform, Object[] -> List<Object>? Narrowing the materialization (e.g. must be a Multimap, by name or by checkArgument) is fine.

A comment on what the method exists for would also be appreciated - my understanding of how this interacts with the existing ViewFns is that they're stored as the encoded objects in a singleton map from the void key; and in the future we'll do something smarter; this will be responsible for materializing those as well?

This is for the future in case we have other materializations then just multimap so we can't give anything more specific then Object.

Your assumption is correct. This is to decouple the SDKs representation of how it chooses to materialize the different views with how the runner is testing them.

Once side inputs are being only fetched over the Fn API then this will not be the way to test things as really the Runner should be dealing just with storage and look up and not need to deal with the ViewFn at all.

tgroh · 2017-10-19T19:58:11Z

...a/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java

  // Generate the temporary-file prefix.
-  private PCollectionView<String> createTempFilePrefixView(PCollectionView<String> jobIdView) {
-    return ((PCollection<String>) jobIdView.getPCollection())
+  private PCollectionView<String> createTempFilePrefixView(


Can this change be separated out? It already LGTM, assuming that we don't ever expect to have separate prefixes per window.

Like part of another PR or just called out as a commit in this PR?

Currently just split to separate commit.

Yeah, that's fine.

tgroh · 2017-10-19T20:48:41Z

runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryMultimapSideInputView.java

+    //  * null keys
+    //  * null values
+    //  * duplicate values
+    Multimap<Object, Object> multimap = ArrayListMultimap.create();


This can be a ListMultimap to force a bit more of the preconditions. Up to you.

The List part isn't important. Its just the ability to store null keys and values and duplicate values. For example HashMultimap doesn't support duplicate values and ImmutableMultimap doesn't support null keys/values.

I wanted to use ListMultimap instead of generic Multimap to signal the non-set nature of the Multimap, as there's unfortunately no BagMultimap, but It's a minor nit given the commenting.

tgroh · 2017-10-19T20:50:07Z

runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryMultimapSideInputView.java

I don't think the firstNotNull is necessary; Guava Multimaps return empty collections when there are no values associated with a key.

Thanks, updated.

tgroh · 2017-10-19T20:52:24Z

runners/core-java/src/main/java/org/apache/beam/runners/core/SideInputHandler.java

-    return sideInput.getViewFn().apply(elements);
+    ViewFn<MultimapView, T> viewFn = (ViewFn<MultimapView, T>) view.getViewFn();
+    Coder<?> keyCoder = ((KvCoder<?, ?>) view.getCoderInternal()).getKeyCoder();
+    return viewFn.apply(InMemoryMultimapSideInputView.fromIterable(keyCoder, (Iterable) elements));


Can/do we have an outstanding task to update the handler to use a more appropriate storage type, and/or is there a more appropriate state type?

The issue from swapping away from value state is that not all runners may support more complex state types and value state effectively forces using an in memory representation.

tgroh · 2017-10-19T20:56:28Z

runners/direct-java/src/main/java/org/apache/beam/runners/direct/SideInputContainer.java

+      @SuppressWarnings("unchecked") Iterable<KV<?, ?>> elements = Iterables.transform(
+          (Iterable<WindowedValue<KV<?, ?>>>) viewContents.getUnchecked(
+              PCollectionViewWindow.of(view, window)).get(),
+          new Function<WindowedValue<KV<?, ?>>, KV<?, ?>>() {


Instead of storing the input Multimap's entries as an iterable and converting back, can we just store the multimap directly? If that's too large for this change, file a followup jira.

The plan was to eventually improve these representations (or have runner authors make these changes), see https://issues.apache.org/jira/browse/BEAM-3080 for the DirectRunner/ULR.

This was just to get the change in since all the runners here just do a trivial multimap -> iterable -> multimap for all the types anyways.

tgroh · 2017-10-19T21:00:37Z

.../src/main/java/org/apache/beam/runners/flink/translation/functions/SideInputInitializer.java

-public class SideInputInitializer<ElemT, ViewT, W extends BoundedWindow>
-    implements BroadcastVariableInitializer<WindowedValue<ElemT>, Map<BoundedWindow, ViewT>> {
+public class SideInputInitializer<ViewT>
+    implements BroadcastVariableInitializer<WindowedValue<?>, Map<BoundedWindow, ViewT>> {


Can this be WindowedValues of KVs, or is that not possible with the module's type signatures?

I'm specifically not narrowing the type because KV<> is for the multimap materialization. When/if this SideInputInitializer supports a different URN, that KV is unlikely to be true anymore.

tgroh · 2017-10-19T21:00:56Z

runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/DoFnOperatorTest.java

Same with a @Before method

tgroh · 2017-10-19T21:01:16Z

...flow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java

No action required, but these number changes are a real downer.

tgroh · 2017-10-19T21:02:53Z

runners/spark/src/main/java/org/apache/beam/runners/spark/util/SparkSideInputReader.java

(nit) can we just use contains?

lukecwik · 2017-10-20T16:10:13Z

Run Java PreCommit

tgroh

LGTM

tgroh · 2017-10-21T00:11:10Z

runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryMultimapSideInputView.java

+    //  * null keys
+    //  * null values
+    //  * duplicate values
+    Multimap<Object, Object> multimap = ArrayListMultimap.create();


I wanted to use ListMultimap instead of generic Multimap to signal the non-set nature of the Multimap, as there's unfortunately no BagMultimap, but It's a minor nit given the commenting.

tgroh · 2017-10-21T00:11:38Z

...a/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java

  // Generate the temporary-file prefix.
-  private PCollectionView<String> createTempFilePrefixView(PCollectionView<String> jobIdView) {
-    return ((PCollection<String>) jobIdView.getPCollection())
+  private PCollectionView<String> createTempFilePrefixView(


Yeah, that's fine.

…n the Java SDK.

lukecwik · 2017-11-13T21:10:28Z

Run Dataflow ValidatesRunner

lukecwik · 2017-11-13T21:10:35Z

Run Apex ValidatesRunner

lukecwik · 2017-11-13T21:10:41Z

Run Spark ValidatesRunner

lukecwik · 2017-11-13T21:10:49Z

Run Flink ValidatesRunner

lukecwik · 2017-11-13T21:10:57Z

Run Gearpump ValidatesRunner

asfgit · 2017-11-13T21:13:08Z

SUCCESS

--none--

asfgit · 2017-11-13T21:22:48Z

SUCCESS

--none--

asfgit · 2017-11-13T21:23:57Z

SUCCESS

--none--

asfgit · 2017-11-13T21:33:59Z

SUCCESS

--none--

asfgit · 2017-11-13T21:34:11Z

SUCCESS

--none--

asfgit · 2017-11-13T21:37:28Z

FAILURE

--none--

asfgit · 2017-11-13T22:07:03Z

FAILURE

--none--

asfgit · 2017-11-13T23:04:06Z

FAILURE

--none--

…implementation is specialized.

lukecwik · 2017-11-14T19:38:38Z

Run Dataflow ValidatesRunner

lukecwik · 2017-11-14T19:38:46Z

Run Spark ValidatesRunner

lukecwik · 2017-11-14T19:38:55Z

Run Apex ValidatesRunner

lukecwik · 2017-11-14T19:39:04Z

Run Gearpump ValidatesRunner

lukecwik · 2017-11-14T19:39:12Z

Run Flink ValidatesaRunner

asfgit · 2017-11-14T19:41:07Z

SUCCESS

--none--

lukecwik · 2017-11-14T19:41:28Z

Run Flink ValidatesRunner

asfgit · 2017-11-14T19:48:41Z

SUCCESS

--none--

asfgit · 2017-11-14T19:50:36Z

FAILURE

--none--

asfgit · 2017-11-14T19:52:58Z

SUCCESS

--none--

asfgit · 2017-11-14T20:00:21Z

SUCCESS

--none--

asfgit · 2017-11-14T20:04:48Z

SUCCESS

--none--

asfgit · 2017-11-14T20:17:07Z

FAILURE

--none--

asfgit · 2017-11-14T20:17:52Z

FAILURE

--none--

…as being replaced.

asfgit · 2017-11-15T03:23:32Z

SUCCESS

--none--

lukecwik · 2017-11-15T03:24:41Z

Run Dataflow ValidatesRunner

lukecwik · 2017-11-15T03:24:48Z

Run Apex ValidatesRunner

lukecwik · 2017-11-15T03:24:55Z

Run Spark ValidatesRunner

lukecwik · 2017-11-15T03:25:15Z

Run Flink ValidatesRunner

lukecwik · 2017-11-15T03:25:30Z

Run Gearpump ValidatesRunner

asfgit · 2017-11-15T03:34:31Z

SUCCESS

--none--

asfgit · 2017-11-15T03:39:43Z

SUCCESS

--none--

asfgit · 2017-11-15T03:41:00Z

SUCCESS

--none--

asfgit · 2017-11-15T03:46:22Z

SUCCESS

--none--

asfgit · 2017-11-15T03:47:24Z

SUCCESS

--none--

asfgit · 2017-11-15T04:23:43Z

SUCCESS

--none--

asfgit · 2017-11-15T06:45:45Z

SUCCESS

--none--

lukecwik · 2017-11-15T16:58:20Z

@tgroh I'm going to merge this as is to avoid the continued Dataflow worker dance.

Feel free to add additional comments for 341521a and I'll apply them in a follow up PR.

lukecwik force-pushed the side_input branch 2 times, most recently from f9dff19 to f7240d8 Compare October 19, 2017 16:32

tgroh reviewed Oct 19, 2017

View reviewed changes

tgroh approved these changes Oct 21, 2017

View reviewed changes

lukecwik force-pushed the side_input branch 3 times, most recently from ba39ffd to 76c4fc0 Compare October 25, 2017 21:02

lukecwik added 2 commits November 10, 2017 11:08

[BEAM-2926] Migrate to using a trivial multimap materialization withi…

6f91ea1

…n the Java SDK.

Rebase and update code to honor findbugs @Nullable/@nonnull conversion

1be1166

lukecwik force-pushed the side_input branch from 76c4fc0 to 1be1166 Compare November 10, 2017 19:31

Update Dataflow worker image to be compatible with side input changes.

0e903a1

Replace the View.As transforms for Dataflow batch because the entire …

6901b8f

…implementation is specialized.

Fix BatchViewOverrides ViewAsSingleton to apply the combine fn that w…

341521a

…as being replaced.

asfgit closed this in 30886ac Nov 15, 2017

kennknowles mentioned this pull request Oct 15, 2020

WIP: Resolve beam_fn_api experiment for side inputs, using portable expansion always #13053

Closed

4 tasks

kennknowles mentioned this pull request Jun 3, 2022

Improve the side input materialization for the DirectRunner/ULR from iterable to storing the multimap directly #18517

Open

Conversation

lukecwik commented Oct 18, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

lukecwik commented Oct 19, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lukecwik Oct 19, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lukecwik Oct 19, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lukecwik commented Oct 20, 2017

Uh oh!

tgroh left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lukecwik commented Nov 13, 2017

Uh oh!

lukecwik commented Nov 13, 2017

Uh oh!

lukecwik commented Nov 13, 2017

Uh oh!

lukecwik commented Nov 13, 2017

Uh oh!

lukecwik commented Nov 13, 2017

Uh oh!

asfgit commented Nov 13, 2017

Uh oh!

asfgit commented Nov 13, 2017

Uh oh!

asfgit commented Nov 13, 2017

Uh oh!

asfgit commented Nov 13, 2017

Uh oh!

asfgit commented Nov 13, 2017

lukecwik commented Oct 18, 2017 •

edited

Loading

lukecwik Oct 19, 2017 •

edited

Loading

lukecwik Oct 19, 2017 •

edited

Loading