[
https://issues.apache.org/jira/browse/BEAM-4271?focusedWorklogId=103510&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-103510
]
ASF GitHub Bot logged work on BEAM-4271:
----------------------------------------
Author: ASF GitHub Bot
Created on: 18/May/18 18:06
Start Date: 18/May/18 18:06
Worklog Time Spent: 10m
Work Description: bsidhom commented on a change in pull request #5374:
[BEAM-4271] Support side inputs for ExecutableStage and provide runner side
utilities for handling multimap side inputs.
URL: https://github.com/apache/beam/pull/5374#discussion_r189349159
##########
File path:
runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/state/StateRequestHandlers.java
##########
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.beam.runners.fnexecution.state;
+
+import static com.google.common.base.Preconditions.checkState;
+
+import com.google.protobuf.ByteString;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.StateGetResponse;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.StateKey;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.StateKey.TypeCase;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.StateRequest;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.StateResponse;
+import
org.apache.beam.runners.fnexecution.control.ProcessBundleDescriptors.ExecutableProcessBundleDescriptor;
+import
org.apache.beam.runners.fnexecution.control.ProcessBundleDescriptors.MultimapSideInputSpec;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.fn.stream.DataStreams;
+import org.apache.beam.sdk.fn.stream.DataStreams.ElementDelimitedOutputStream;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.util.common.Reiterable;
+
+/**
+ * A set of utility methods which construct {@link StateRequestHandler}s.
+ *
+ * <p>TODO: Add a variant which works on {@link ByteString}s to remove
encoding/decoding overhead.
+ */
+public class StateRequestHandlers {
+
+ /**
+ * A handler for multimap side inputs.
+ */
+ public interface MultimapSideInputHandler<K, V, W extends BoundedWindow> {
+ /**
+ * Returns an {@link Iterable} of values representing the side input for
the given key and
+ * window.
+ *
+ * <p>TODO: Add support for side input chunking and caching if a {@link
Reiterable} is returned.
+ */
+ Iterable<V> get(K key, W window);
+ }
+
+ /**
+ * A factory which constructs {@link MultimapSideInputHandler}s.
+ */
+ public interface MultimapSideInputHandlerFactory {
+
+ /**
+ * Returns a {@link MultimapSideInputHandler} for the given {@code
pTransformId} and
+ * {@code sideInputId}. The supplied {@code keyCoder}, {@code valueCoder},
and
+ * {@code windowCoder} should be used to encode/decode their respective
values.
+ */
+ <K, V, W extends BoundedWindow> MultimapSideInputHandler<K, V, W>
forSideInput(
+ String pTransformId,
+ String sideInputId,
+ Coder<K> keyCoder,
+ Coder<V> valueCoder,
+ Coder<W> windowCoder);
+
+ /**
+ * Throws a {@link UnsupportedOperationException} on the first access.
+ */
+ static MultimapSideInputHandlerFactory unsupported() {
+ return new MultimapSideInputHandlerFactory() {
+ @Override
+ public <K, V, W extends BoundedWindow> MultimapSideInputHandler<K, V,
W> forSideInput(
+ String pTransformId, String sideInputId, Coder<K> keyCoder,
Coder<V> valueCoder,
+ Coder<W> windowCoder) {
+ throw new UnsupportedOperationException(String.format(
+ "The %s does not support handling sides inputs for PTransform %s
with side "
+ + "input id %s.",
+ MultimapSideInputHandler.class.getSimpleName(),
+ pTransformId,
+ sideInputId));
+ }
+ };
+ }
+ }
+
+ /**
+ * A handler for bag user state.
+ */
+ public interface BagUserStateHandler<K, V, W extends BoundedWindow> {
+ /**
+ * Returns an {@link Iterable} of values representing the bag user state
for the given key and
+ * window.
+ *
+ * <p>TODO: Add support for bag user state chunking and caching if a
{@link Reiterable} is
+ * returned.
+ */
+ Iterable<V> get(K key, W window);
+
+ /**
+ * Appends the values to the bag user state for the given key and window.
+ */
+ void append(K key, W window, Iterator<V> values);
+
+ /**
+ * Clears the bag user state for the given key and window.
+ */
+ void clear(K key, W window);
+ }
+
+ /**
+ * A factory which constructs {@link BagUserStateHandler}s.
+ */
+ public interface BagUserStateHandlerFactory {
+ <K, V, W extends BoundedWindow> BagUserStateHandler<K, V, W> forUserState(
+ String pTransformId,
+ String userStateId,
+ Coder<K> keyCoder,
+ Coder<V> valueCoder,
+ Coder<W> windowCoder);
+
+ /**
+ * Throws a {@link UnsupportedOperationException} on the first access.
+ */
+ static BagUserStateHandlerFactory unsupported() {
+ return new BagUserStateHandlerFactory() {
+ @Override
+ public <K, V, W extends BoundedWindow> BagUserStateHandler<K, V, W>
forUserState(
+ String pTransformId, String userStateId, Coder<K> keyCoder,
Coder<V> valueCoder,
+ Coder<W> windowCoder) {
+ throw new UnsupportedOperationException(String.format(
+ "The %s does not support handling sides inputs for PTransform %s
with user state "
+ + "id %s.",
+ BagUserStateHandler.class.getSimpleName(),
+ pTransformId,
+ userStateId));
+ }
+ };
+ }
+ }
+
+ /**
+ * Returns an adapter which converts a {@link
MultimapSideInputHandlerFactory} to
+ * {@link StateRequestHandler}.
+ *
+ * <p>The {@link MultimapSideInputHandlerFactory} is required to handle all
multimap side inputs
+ * contained within the {@link ExecutableProcessBundleDescriptor}. See
+ * {@link ExecutableProcessBundleDescriptor#getMultimapSideInputSpecs} for
the set of multimap
+ * side inputs that are contained.
+ */
+ public static StateRequestHandler forMultimapSideInputHandlerFactory(
+ ExecutableProcessBundleDescriptor processBundleDescriptor,
+ MultimapSideInputHandlerFactory multimapSideInputHandlerFactory) {
+ return new StateRequestHandlerToMultimapSideInputHandlerFactoryAdapter(
+ processBundleDescriptor, multimapSideInputHandlerFactory);
+ }
+
+ /**
+ * An adapter which converts {@link MultimapSideInputHandlerFactory} to
+ * {@link StateRequestHandler}.
+ */
+ static class StateRequestHandlerToMultimapSideInputHandlerFactoryAdapter
+ implements StateRequestHandler {
+
+ private final ExecutableProcessBundleDescriptor processBundleDescriptor;
+ private final MultimapSideInputHandlerFactory
multimapSideInputHandlerFactory;
+ private final ConcurrentHashMap<MultimapSideInputSpec,
MultimapSideInputHandler> cache;
+
+ StateRequestHandlerToMultimapSideInputHandlerFactoryAdapter(
+ ExecutableProcessBundleDescriptor processBundleDescriptor,
+ MultimapSideInputHandlerFactory multimapSideInputHandlerFactory) {
+ this.processBundleDescriptor = processBundleDescriptor;
+ this.multimapSideInputHandlerFactory = multimapSideInputHandlerFactory;
+ this.cache = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public CompletionStage<StateResponse.Builder> handle(
+ StateRequest request) throws Exception {
+ try {
+
checkState(TypeCase.MULTIMAP_SIDE_INPUT.equals(request.getStateKey().getTypeCase()),
+ "Unsupported %s type %s, expected %s",
+ StateRequest.class.getSimpleName(),
+ request.getStateKey().getTypeCase(),
+ TypeCase.MULTIMAP_SIDE_INPUT);
+
+ StateKey.MultimapSideInput stateKey =
request.getStateKey().getMultimapSideInput();
+ MultimapSideInputSpec<?, ?, ?> sideInputReferenceSpec =
+ processBundleDescriptor.getMultimapSideInputSpecs()
+ .get(stateKey.getPtransformId())
+ .get(stateKey.getSideInputId());
+ MultimapSideInputHandler<?, ?, ?> handler = cache.computeIfAbsent(
Review comment:
Note that in general, side input handlers may have limited lifetimes and
will be invalid if referenced outside of that lifetime. Should we provide a way
to explicitly remove entries from the cache and/or mark them as inoperable?
In either case, please document the fact that the
`MultimapSideInputHandler`s returned by factories must be thread-safe.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on 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: 103510)
Time Spent: 1h (was: 50m)
> Executable stages allow side input coders to be set and/or queried
> ------------------------------------------------------------------
>
> Key: BEAM-4271
> URL: https://issues.apache.org/jira/browse/BEAM-4271
> Project: Beam
> Issue Type: Bug
> Components: runner-core
> Reporter: Ben Sidhom
> Assignee: Luke Cwik
> Priority: Major
> Time Spent: 1h
> Remaining Estimate: 0h
>
> ProcessBundleDescriptors may contain side input references from inner
> PTransforms. These side inputs do not have explicit coders; instead, SDK
> harnesses use the PCollection coders by default.
> Using the default PCollection coder as specified at pipeline construction is
> in general not the correct thing to do. When PCollection elements are
> materialized, any coders unknown to a runner a length-prefixed. This means
> that materialized PCollections do not use their original element coders. Side
> inputs are delivered to SDKs via MultimapSideInput StateRequests. The
> responses to these requests are expected to contain all of the values for a
> given key (and window), coded with the PCollection KV.value coder,
> concatenated. However, at the time of serving these requests on the runner
> side, we do not have enough information to reconstruct the original value
> coders.
> There are different ways to address this issue. For example:
> * Modify the associated PCollection coder to match the coder that the runner
> uses to materialize elements. This means that anywhere a given PCollection is
> used within a given bundle, it will use the runner-safe coder. This may
> introduce inefficiencies but should be "correct".
> * Annotate side inputs with explicit coders. This guarantees that the key
> and value coders used by the runner match the coders used by SDKs.
> Furthermore, it allows the _runners_ to specify coders. This involves changes
> to the proto models and all SDKs.
> * Annotate side input state requests with both key and value coders. This
> inverts the expected responsibility and has the SDK determine runner coders.
> Additionally, because runners do not understand all SDK types, additional
> coder substitution will need to be done at request handling time to make sure
> that the requested coder can be instantiated and will remain consistent with
> the SDK coder. This requires only small changes to SDKs because they may opt
> to use their default PCollection coders.
> All of the these approaches have their own downsides. Explicit side input
> coders is probably the right thing to do long-term, but the simplest change
> for now is to modify PCollection coders to match exactly how they're
> materialized.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
