beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (BEAM-2863) Add support for Side Inputs over the Fn API
Date Fri, 12 Jan 2018 22:37:00 GMT

    [ https://issues.apache.org/jira/browse/BEAM-2863?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16324674#comment-16324674
] 

ASF GitHub Bot commented on BEAM-2863:
--------------------------------------

lukecwik closed pull request #4377: [BEAM-2863] Add the ability to length prefix unknown coders
URL: https://github.com/apache/beam/pull/4377
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/runners/java-fn-execution/build.gradle b/runners/java-fn-execution/build.gradle
index e948c7ce625..9bfa32c63f1 100644
--- a/runners/java-fn-execution/build.gradle
+++ b/runners/java-fn-execution/build.gradle
@@ -40,6 +40,7 @@ dependencies {
   shadow library.java.grpc_netty
   shadow library.java.slf4j_api
   testCompile project(":beam-sdks-parent:beam-sdks-java-parent:beam-sdks-java-fn-execution").sourceSets.test.output
+  testCompile project(path: ":beam-runners-parent:beam-runners-core-construction-java", configuration:
"shadow")
   testCompile library.java.junit
   testCompile library.java.hamcrest_core
   testCompile library.java.hamcrest_library
diff --git a/runners/java-fn-execution/pom.xml b/runners/java-fn-execution/pom.xml
index 1877e2a56ab..77f1d317e6e 100644
--- a/runners/java-fn-execution/pom.xml
+++ b/runners/java-fn-execution/pom.xml
@@ -97,6 +97,12 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>org.apache.beam</groupId>
+      <artifactId>beam-runners-core-construction-java</artifactId>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>org.apache.beam</groupId>
       <artifactId>beam-sdks-java-fn-execution</artifactId>
diff --git a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/graph/LengthPrefixUnknownCoders.java
b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/graph/LengthPrefixUnknownCoders.java
new file mode 100644
index 00000000000..ac7e745b939
--- /dev/null
+++ b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/graph/LengthPrefixUnknownCoders.java
@@ -0,0 +1,183 @@
+/*
+ * Copyright (C) 2017 Google Inc.
+ *
+ * Licensed 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.graph;
+
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Sets;
+import java.util.Set;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.model.pipeline.v1.RunnerApi.Coder;
+import org.apache.beam.model.pipeline.v1.RunnerApi.Components;
+import org.apache.beam.model.pipeline.v1.RunnerApi.MessageWithComponents;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.LengthPrefixCoder;
+
+/**
+ * Utilities for replacing or wrapping unknown coders with {@link LengthPrefixCoder}.
+ *
+ * <p>TODO: Support a dynamic list of well known coders using either registration or
manual listing.
+ */
+public class LengthPrefixUnknownCoders {
+  private static final String BYTES_CODER_TYPE = "urn:beam:coders:bytes:0.1";
+  private static final String LENGTH_PREFIX_CODER_TYPE = "urn:beam:coders:length_prefix:0.1";
+  private static final Set<String> WELL_KNOWN_CODER_URNS =
+      ImmutableSet.of(
+          BYTES_CODER_TYPE,
+          "urn:beam:coders:kv:0.1",
+          "urn:beam:coders:varint:0.1",
+          "urn:beam:coders:interval_window:0.1",
+          "urn:beam:coders:stream:0.1",
+          LENGTH_PREFIX_CODER_TYPE,
+          "urn:beam:coders:global_window:0.1",
+          "urn:beam:coders:windowed_value:0.1");
+
+  /**
+   * Recursively traverse the coder tree and wrap the first unknown coder in every branch
with a
+   * {@link LengthPrefixCoder} unless an ancestor coder is itself a {@link LengthPrefixCoder}.
If
+   * {@code replaceWithByteArrayCoder} is set, then replace that unknown coder with a
+   * {@link ByteArrayCoder}. Note that no ids that are generated will collide with the ids
supplied
+   * within the {@link Components#getCodersMap() coder map} key space.
+   *
+   * @param coderId The root coder contained within {@code coders} to start the recursive
descent
+   * from.
+   * @param components Contains the root coder and all component coders.
+   * @param replaceWithByteArrayCoder whether to replace an unknown coder with a
+   * {@link ByteArrayCoder}.
+   * @return A {@link MessageWithComponents} with the
+   * {@link MessageWithComponents#getCoder() root coder} and its component coders. Note that
no ids
+   * that are generated will collide with the ids supplied within the
+   * {@link Components#getCodersMap() coder map} key space.
+   */
+  public static RunnerApi.MessageWithComponents forCoder(
+      String coderId,
+      RunnerApi.Components components,
+      boolean replaceWithByteArrayCoder) {
+
+    RunnerApi.Coder currentCoder = components.getCodersOrThrow(coderId);
+
+    // We handle three cases:
+    //  1) the requested coder is already a length prefix coder. In this case we just honor
the
+    //     request to replace the coder with a byte array coder.
+    //  2) the requested coder is a known coder but not a length prefix coder. In this case
we
+    //     rebuild the coder by recursively length prefixing any unknown component coders.
+    //  3) the requested coder is an unknown coder. In this case we either wrap the requested
coder
+    //     with a length prefix coder or replace it with a length prefix byte array coder.
+    if (LENGTH_PREFIX_CODER_TYPE.equals(currentCoder.getSpec().getSpec().getUrn())) {
+      if (replaceWithByteArrayCoder) {
+        return createLengthPrefixByteArrayCoder(coderId, components);
+      }
+
+      MessageWithComponents.Builder rvalBuilder = MessageWithComponents.newBuilder();
+      rvalBuilder.setCoder(currentCoder);
+      rvalBuilder.setComponents(components);
+      return rvalBuilder.build();
+    } else if (WELL_KNOWN_CODER_URNS.contains(currentCoder.getSpec().getSpec().getUrn()))
{
+      return lengthPrefixUnknownComponentCoders(coderId, components, replaceWithByteArrayCoder);
+    } else {
+      return lengthPrefixUnknownCoder(coderId, components, replaceWithByteArrayCoder);
+    }
+  }
+
+  private static MessageWithComponents lengthPrefixUnknownComponentCoders(
+      String coderId,
+      RunnerApi.Components components,
+      boolean replaceWithByteArrayCoder) {
+
+    MessageWithComponents.Builder rvalBuilder = MessageWithComponents.newBuilder();
+    RunnerApi.Coder currentCoder = components.getCodersOrThrow(coderId);
+    RunnerApi.Coder.Builder updatedCoder = currentCoder.toBuilder();
+    // Rebuild the component coder ids to handle if any of the component coders changed.
+    updatedCoder.clearComponentCoderIds();
+    for (final String componentCoderId : currentCoder.getComponentCoderIdsList()) {
+      MessageWithComponents componentCoder =
+          forCoder(componentCoderId, components, replaceWithByteArrayCoder);
+      String newComponentCoderId = componentCoderId;
+      if (!components.getCodersOrThrow(componentCoderId).equals(componentCoder.getCoder()))
{
+        // Generate a new id if the component coder changed.
+        newComponentCoderId = generateUniqueId(
+            coderId + "-length_prefix",
+            Sets.union(components.getCodersMap().keySet(),
+                rvalBuilder.getComponents().getCodersMap().keySet()));
+      }
+      updatedCoder.addComponentCoderIds(newComponentCoderId);
+      rvalBuilder.getComponentsBuilder().putCoders(newComponentCoderId, componentCoder.getCoder());
+      // Insert all component coders of the component coder.
+      rvalBuilder.getComponentsBuilder().putAllCoders(
+          componentCoder.getComponents().getCodersMap());
+    }
+    rvalBuilder.setCoder(updatedCoder);
+
+    return rvalBuilder.build();
+  }
+
+  // If we are handling an unknown URN then we need to wrap it with a length prefix coder.
+  // If requested we also replace the unknown coder with a byte array coder.
+  private static MessageWithComponents lengthPrefixUnknownCoder(
+      String coderId,
+      RunnerApi.Components components,
+      boolean replaceWithByteArrayCoder) {
+    MessageWithComponents.Builder rvalBuilder = MessageWithComponents.newBuilder();
+    RunnerApi.Coder currentCoder = components.getCodersOrThrow(coderId);
+
+    String lengthPrefixComponentCoderId = coderId;
+    if (replaceWithByteArrayCoder) {
+      return createLengthPrefixByteArrayCoder(coderId, components);
+    } else {
+      rvalBuilder.getComponentsBuilder().putCoders(coderId, currentCoder);
+    }
+
+    rvalBuilder.getCoderBuilder()
+        .addComponentCoderIds(lengthPrefixComponentCoderId)
+        .getSpecBuilder()
+        .getSpecBuilder()
+        .setUrn(LENGTH_PREFIX_CODER_TYPE);
+    return rvalBuilder.build();
+  }
+
+  private static MessageWithComponents createLengthPrefixByteArrayCoder(
+      String coderId,
+      RunnerApi.Components components) {
+    MessageWithComponents.Builder rvalBuilder = MessageWithComponents.newBuilder();
+
+    String byteArrayCoderId = generateUniqueId(
+        coderId + "-byte_array",
+        Sets.union(components.getCodersMap().keySet(),
+            rvalBuilder.getComponents().getCodersMap().keySet()));
+    Coder.Builder byteArrayCoder = Coder.newBuilder();
+    byteArrayCoder.getSpecBuilder().getSpecBuilder().setUrn(BYTES_CODER_TYPE);
+    rvalBuilder.getComponentsBuilder().putCoders(byteArrayCoderId,
+        byteArrayCoder.build());
+    rvalBuilder.getCoderBuilder()
+        .addComponentCoderIds(byteArrayCoderId)
+        .getSpecBuilder()
+        .getSpecBuilder()
+        .setUrn(LENGTH_PREFIX_CODER_TYPE);
+
+    return rvalBuilder.build();
+  }
+
+  /**
+   * Generates a unique id given a prefix and the set of existing ids.
+   */
+  static String generateUniqueId(String prefix, Set<String> existingIds) {
+    int i = 0;
+    while (existingIds.contains(prefix + i)) {
+      i += 1;
+    }
+    return prefix + i;
+  }
+}
diff --git a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/graph/package-info.java
b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/graph/package-info.java
new file mode 100644
index 00000000000..c7ffe538b63
--- /dev/null
+++ b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/graph/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+/**
+ * Utilities for a Beam runner to interact with the pipeline proto representation.
+ */
+package org.apache.beam.runners.fnexecution.graph;
diff --git a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/graph/LengthPrefixUnknownCodersTest.java
b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/graph/LengthPrefixUnknownCodersTest.java
new file mode 100644
index 00000000000..4cfa5c0c545
--- /dev/null
+++ b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/graph/LengthPrefixUnknownCodersTest.java
@@ -0,0 +1,147 @@
+/*
+ * Copyright (C) 2017 Google Inc.
+ *
+ * Licensed 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.graph;
+
+import static org.junit.Assert.assertEquals;
+
+import com.google.common.collect.ImmutableList;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collection;
+import org.apache.beam.model.pipeline.v1.RunnerApi.Components;
+import org.apache.beam.model.pipeline.v1.RunnerApi.MessageWithComponents;
+import org.apache.beam.runners.core.construction.CoderTranslation;
+import org.apache.beam.runners.core.construction.RehydratedComponents;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.LengthPrefixCoder;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameter;
+import org.junit.runners.Parameterized.Parameters;
+
+/** Tests for {@link LengthPrefixUnknownCoders}. */
+@RunWith(Parameterized.class)
+public class LengthPrefixUnknownCodersTest {
+
+  private static class UnknownCoder extends CustomCoder<String> {
+    private static final Coder<?> INSTANCE = new UnknownCoder();
+    @Override
+    public void encode(String value, OutputStream outStream) throws CoderException, IOException
{
+    }
+
+    @Override
+    public String decode(InputStream inStream) throws CoderException, IOException {
+      return "";
+    }
+
+    @Override
+    public int hashCode() {
+      return 1278890232;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      return obj instanceof  UnknownCoder;
+    }
+  }
+
+  @Parameters
+  public static Collection<Object[]> data() {
+    return ImmutableList.of(
+        /** Test wrapping unknown coders with {@code LengthPrefixCoder}. */
+        new Object[] {
+            WindowedValue.getFullCoder(
+                KvCoder.of(UnknownCoder.INSTANCE, UnknownCoder.INSTANCE),
+                GlobalWindow.Coder.INSTANCE),
+            WindowedValue.getFullCoder(
+                KvCoder.of(LengthPrefixCoder.of(UnknownCoder.INSTANCE),
+                    LengthPrefixCoder.of(UnknownCoder.INSTANCE)),
+                GlobalWindow.Coder.INSTANCE),
+            false
+        },
+        /**
+         * Test bypassing unknown coders that are already wrapped with
+         * {@code LengthPrefixCoder}.
+         */
+        new Object[] {
+            WindowedValue.getFullCoder(
+                KvCoder.of(UnknownCoder.INSTANCE,
+                    LengthPrefixCoder.of(UnknownCoder.INSTANCE)),
+                GlobalWindow.Coder.INSTANCE),
+            WindowedValue.getFullCoder(
+                KvCoder.of(LengthPrefixCoder.of(UnknownCoder.INSTANCE),
+                    LengthPrefixCoder.of(UnknownCoder.INSTANCE)),
+                GlobalWindow.Coder.INSTANCE),
+            false
+        },
+        /** Test replacing unknown coders with {@code LengthPrefixCoder<ByteArray>}.
*/
+        new Object[] {
+            WindowedValue.getFullCoder(
+                KvCoder.of(LengthPrefixCoder.of(UnknownCoder.INSTANCE),
+                    UnknownCoder.INSTANCE),
+                GlobalWindow.Coder.INSTANCE),
+            WindowedValue.getFullCoder(
+                KvCoder.of(LengthPrefixCoder.of(ByteArrayCoder.of()),
+                    LengthPrefixCoder.of(ByteArrayCoder.of())),
+                GlobalWindow.Coder.INSTANCE),
+            true
+        },
+        /** Test skipping a top level length prefix coder. */
+        new Object[] {
+            LengthPrefixCoder.of(UnknownCoder.INSTANCE),
+            LengthPrefixCoder.of(UnknownCoder.INSTANCE),
+            false
+        },
+        /** Test replacing a top level length prefix coder with byte array coder. */
+        new Object[] {
+            LengthPrefixCoder.of(UnknownCoder.INSTANCE),
+            LengthPrefixCoder.of(ByteArrayCoder.of()),
+            true
+        }
+    );
+  }
+
+  @Parameter
+  public Coder<?> original;
+
+  @Parameter(1)
+  public Coder<?> expected;
+
+  @Parameter(2)
+  public boolean replaceWithByteArray;
+
+  @Test
+  public void test() throws IOException {
+    MessageWithComponents originalCoderProto = CoderTranslation.toProto(original);
+    Components.Builder builder = originalCoderProto.getComponents().toBuilder();
+    String coderId = LengthPrefixUnknownCoders.generateUniqueId("rootTestId",
+        originalCoderProto.getComponents().getCodersMap().keySet());
+    builder.putCoders(coderId, originalCoderProto.getCoder());
+    MessageWithComponents updatedCoderProto = LengthPrefixUnknownCoders.forCoder(
+        coderId, builder.build(), replaceWithByteArray);
+    assertEquals(expected,
+        CoderTranslation.fromProto(updatedCoderProto.getCoder(),
+            RehydratedComponents.forComponents(updatedCoderProto.getComponents())));
+  }
+}


 

----------------------------------------------------------------
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:
users@infra.apache.org


> Add support for Side Inputs over the Fn API
> -------------------------------------------
>
>                 Key: BEAM-2863
>                 URL: https://issues.apache.org/jira/browse/BEAM-2863
>             Project: Beam
>          Issue Type: Improvement
>          Components: beam-model
>            Reporter: Luke Cwik
>            Assignee: Luke Cwik
>              Labels: portability
>             Fix For: 2.3.0
>
>
> See:
> * https://s.apache.org/beam-side-inputs-1-pager
> * http://s.apache.org/beam-fn-api-state-api



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

Mime
View raw message