beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dhalp...@apache.org
Subject [1/2] incubator-beam git commit: Deduplicate Unbounded Reads
Date Wed, 22 Jun 2016 16:57:27 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/master 3bc0fe669 -> c83e5c48f


Deduplicate Unbounded Reads

This ensures that sources that produce duplicate elements that are
marked as requiresDeduplication are handled by the DirectRunner.


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/569228e4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/569228e4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/569228e4

Branch: refs/heads/master
Commit: 569228e4c5fc2e3a722d8f3089d75cc6fc197d93
Parents: 3bc0fe6
Author: Thomas Groh <tgroh@google.com>
Authored: Wed Jun 15 14:41:09 2016 -0700
Committer: Dan Halperin <dhalperi@google.com>
Committed: Wed Jun 22 09:57:23 2016 -0700

----------------------------------------------------------------------
 .../direct/UnboundedReadDeduplicator.java       | 102 ++++++++++++++
 .../direct/UnboundedReadEvaluatorFactory.java   |  17 ++-
 .../direct/UnboundedReadDeduplicatorTest.java   | 134 +++++++++++++++++++
 .../UnboundedReadEvaluatorFactoryTest.java      |  50 ++++++-
 4 files changed, 299 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/569228e4/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadDeduplicator.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadDeduplicator.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadDeduplicator.java
new file mode 100644
index 0000000..0246236
--- /dev/null
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadDeduplicator.java
@@ -0,0 +1,102 @@
+/*
+ * 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.direct;
+
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.io.Read.Unbounded;
+import org.apache.beam.sdk.transforms.PTransform;
+
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+
+import org.joda.time.Duration;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Provides methods to determine if a record is a duplicate within the evaluation of a
+ * {@link Unbounded} {@link PTransform}.
+ */
+interface UnboundedReadDeduplicator {
+  /**
+   * Returns true if the record with the provided ID should be output, and false if it should
not
+   * be because it is a duplicate.
+   */
+  boolean shouldOutput(byte[] recordId);
+
+  /**
+   * An {@link UnboundedReadDeduplicator} that always returns true. For use with sources
do not
+   * require deduplication.
+   */
+  class NeverDeduplicator implements UnboundedReadDeduplicator {
+    /**
+     * Create a new {@link NeverDeduplicator}.
+     */
+    public static UnboundedReadDeduplicator create() {
+      return new NeverDeduplicator();
+    }
+
+    @Override
+    public boolean shouldOutput(byte[] recordId) {
+      return true;
+    }
+  }
+
+
+  /**
+   * An {@link UnboundedReadDeduplicator} that returns true if the record ID has not been
seen
+   * within 10 minutes.
+   */
+  class CachedIdDeduplicator implements UnboundedReadDeduplicator {
+    private static final ByteArrayCoder RECORD_ID_CODER = ByteArrayCoder.of();
+    private static final long MAX_RETENTION_SINCE_ACCESS =
+        Duration.standardMinutes(10L).getMillis();
+
+    private final LoadingCache<StructuralKey<byte[]>, AtomicBoolean> ids;
+
+    /**
+     * Create a new {@link CachedIdDeduplicator}.
+     */
+    public static UnboundedReadDeduplicator create() {
+      return new CachedIdDeduplicator();
+    }
+
+    private CachedIdDeduplicator() {
+      ids = CacheBuilder.newBuilder()
+          .expireAfterAccess(MAX_RETENTION_SINCE_ACCESS, TimeUnit.MILLISECONDS)
+          .maximumSize(100_000L)
+          .build(new TrueBooleanLoader());
+    }
+
+    @Override
+    public boolean shouldOutput(byte[] recordId) {
+      return ids.getUnchecked(StructuralKey.of(recordId, RECORD_ID_CODER)).getAndSet(false);
+    }
+
+    private static class TrueBooleanLoader
+        extends CacheLoader<StructuralKey<byte[]>, AtomicBoolean> {
+      @Override
+      public AtomicBoolean load(StructuralKey<byte[]> key) throws Exception {
+        return new AtomicBoolean(true);
+      }
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/569228e4/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java
index 3fb773e..9a287b7 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java
@@ -105,9 +105,15 @@ class UnboundedReadEvaluatorFactory implements TransformEvaluatorFactory
{
         // factory for this transform
         UnboundedSource<OutputT, CheckpointMarkT> source =
             (UnboundedSource<OutputT, CheckpointMarkT>) transform.getTransform().getSource();
+        UnboundedReadDeduplicator deduplicator;
+        if (source.requiresDeduping()) {
+          deduplicator = UnboundedReadDeduplicator.CachedIdDeduplicator.create();
+        } else {
+          deduplicator = UnboundedReadDeduplicator.NeverDeduplicator.create();
+        }
         UnboundedReadEvaluator<OutputT, CheckpointMarkT> evaluator =
             new UnboundedReadEvaluator<>(
-                transform, evaluationContext, source, evaluatorQueue);
+                transform, evaluationContext, source, deduplicator, evaluatorQueue);
         evaluatorQueue.offer(evaluator);
       } else {
         // otherwise return the existing Queue that arrived before us
@@ -142,6 +148,7 @@ class UnboundedReadEvaluatorFactory implements TransformEvaluatorFactory
{
      * source as derived from {@link #transform} due to splitting.
      */
     private final UnboundedSource<OutputT, CheckpointMarkT> source;
+    private final UnboundedReadDeduplicator deduplicator;
     private UnboundedReader<OutputT> currentReader;
     private CheckpointMarkT checkpointMark;
 
@@ -155,12 +162,14 @@ class UnboundedReadEvaluatorFactory implements TransformEvaluatorFactory
{
         AppliedPTransform<?, PCollection<OutputT>, Unbounded<OutputT>>
transform,
         EvaluationContext evaluationContext,
         UnboundedSource<OutputT, CheckpointMarkT> source,
+        UnboundedReadDeduplicator deduplicator,
         ConcurrentLinkedQueue<UnboundedReadEvaluator<OutputT, CheckpointMarkT>>
evaluatorQueue) {
       this.transform = transform;
       this.evaluationContext = evaluationContext;
       this.evaluatorQueue = evaluatorQueue;
       this.source = source;
       this.currentReader = null;
+      this.deduplicator = deduplicator;
       this.checkpointMark = null;
     }
 
@@ -177,8 +186,10 @@ class UnboundedReadEvaluatorFactory implements TransformEvaluatorFactory
{
         if (elementAvailable) {
           int numElements = 0;
           do {
-            output.add(WindowedValue.timestampedValueInGlobalWindow(currentReader.getCurrent(),
-                currentReader.getCurrentTimestamp()));
+            if (deduplicator.shouldOutput(currentReader.getCurrentRecordId())) {
+              output.add(WindowedValue.timestampedValueInGlobalWindow(currentReader.getCurrent(),
+                  currentReader.getCurrentTimestamp()));
+            }
             numElements++;
           } while (numElements < ARBITRARY_MAX_ELEMENTS && currentReader.advance());
           watermark = currentReader.getWatermark();

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/569228e4/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadDeduplicatorTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadDeduplicatorTest.java
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadDeduplicatorTest.java
new file mode 100644
index 0000000..b3c9012
--- /dev/null
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadDeduplicatorTest.java
@@ -0,0 +1,134 @@
+/*
+ * 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.direct;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+
+import org.apache.beam.runners.direct.UnboundedReadDeduplicator.CachedIdDeduplicator;
+import org.apache.beam.runners.direct.UnboundedReadDeduplicator.NeverDeduplicator;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Tests for {@link UnboundedReadDeduplicator}.
+ */
+@RunWith(JUnit4.class)
+public class UnboundedReadDeduplicatorTest {
+  @Test
+  public void neverDeduplicatorAlwaysTrue() {
+    byte[] id = new byte[] {-1, 2, 4, 22};
+    UnboundedReadDeduplicator dedupper = NeverDeduplicator.create();
+
+    assertThat(dedupper.shouldOutput(id), is(true));
+    assertThat(dedupper.shouldOutput(id), is(true));
+  }
+
+  @Test
+  public void cachedIdDeduplicatorTrueForFirstIdThenFalse() {
+    byte[] id = new byte[] {-1, 2, 4, 22};
+    UnboundedReadDeduplicator dedupper = CachedIdDeduplicator.create();
+
+    assertThat(dedupper.shouldOutput(id), is(true));
+    assertThat(dedupper.shouldOutput(id), is(false));
+  }
+
+  @Test
+  public void cachedIdDeduplicatorMultithreaded() throws InterruptedException {
+    byte[] id = new byte[] {-1, 2, 4, 22};
+    UnboundedReadDeduplicator dedupper = CachedIdDeduplicator.create();
+    final CountDownLatch startSignal = new CountDownLatch(1);
+    int numThreads = 1000;
+    final CountDownLatch readyLatch = new CountDownLatch(numThreads);
+    final CountDownLatch finishLine = new CountDownLatch(numThreads);
+
+    ExecutorService executor = Executors.newCachedThreadPool();
+    AtomicInteger successCount = new AtomicInteger();
+    AtomicInteger failureCount = new AtomicInteger();
+    for (int i = 0; i < numThreads; i++) {
+      executor.submit(new TryOutputIdRunnable(dedupper,
+          id,
+          successCount,
+          failureCount,
+          readyLatch,
+          startSignal,
+          finishLine));
+    }
+
+    readyLatch.await();
+    startSignal.countDown();
+    finishLine.await(10L, TimeUnit.SECONDS);
+    executor.shutdownNow();
+
+    assertThat(successCount.get(), equalTo(1));
+    assertThat(failureCount.get(), equalTo(numThreads - 1));
+  }
+
+  private static class TryOutputIdRunnable implements Runnable {
+    private final UnboundedReadDeduplicator deduplicator;
+    private final byte[] id;
+    private final AtomicInteger successCount;
+    private final AtomicInteger failureCount;
+    private final CountDownLatch readyLatch;
+    private final CountDownLatch startSignal;
+    private final CountDownLatch finishLine;
+
+    public TryOutputIdRunnable(
+        UnboundedReadDeduplicator dedupper,
+        byte[] id,
+        AtomicInteger successCount,
+        AtomicInteger failureCount,
+        CountDownLatch readyLatch,
+        CountDownLatch startSignal,
+        CountDownLatch finishLine) {
+      this.deduplicator = dedupper;
+      this.id = id;
+      this.successCount = successCount;
+      this.failureCount = failureCount;
+      this.readyLatch = readyLatch;
+      this.startSignal = startSignal;
+      this.finishLine = finishLine;
+    }
+
+    @Override
+    public void run() {
+      readyLatch.countDown();
+      try {
+        startSignal.await();
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        throw new RuntimeException(e);
+      }
+      if (deduplicator.shouldOutput(id)) {
+        successCount.incrementAndGet();
+      } else {
+        failureCount.incrementAndGet();
+      }
+      finishLine.countDown();
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/569228e4/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java
index e182e8d..839badf 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java
@@ -40,6 +40,7 @@ import org.apache.beam.sdk.testing.TestPipeline;
 import org.apache.beam.sdk.transforms.AppliedPTransform;
 import org.apache.beam.sdk.transforms.SerializableFunction;
 import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.util.CoderUtils;
 import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.values.PCollection;
 
@@ -138,6 +139,38 @@ public class UnboundedReadEvaluatorFactoryTest {
   }
 
   @Test
+  public void unboundedSourceWithDuplicatesMultipleCalls() throws Exception {
+    Long[] outputs = new Long[20];
+    for (long i = 0L; i < 20L; i++) {
+      outputs[(int) i] = i % 5L;
+    }
+    TestUnboundedSource<Long> source =
+        new TestUnboundedSource<>(BigEndianLongCoder.of(), outputs);
+    source.dedupes = true;
+
+    TestPipeline p = TestPipeline.create();
+    PCollection<Long> pcollection = p.apply(Read.from(source));
+    AppliedPTransform<?, ?, ?> sourceTransform = pcollection.getProducingTransformInternal();
+
+    UncommittedBundle<Long> output = bundleFactory.createRootBundle(pcollection);
+    when(context.createRootBundle(pcollection)).thenReturn(output);
+    TransformEvaluator<?> evaluator = factory.forApplication(sourceTransform, null,
context);
+
+    evaluator.finishBundle();
+    assertThat(
+        output.commit(Instant.now()).getElements(),
+        containsInAnyOrder(tgw(1L), tgw(2L), tgw(4L), tgw(3L), tgw(0L)));
+
+    UncommittedBundle<Long> secondOutput = bundleFactory.createRootBundle(longs);
+    when(context.createRootBundle(longs)).thenReturn(secondOutput);
+    TransformEvaluator<?> secondEvaluator = factory.forApplication(sourceTransform,
null, context);
+    secondEvaluator.finishBundle();
+    assertThat(
+        secondOutput.commit(Instant.now()).getElements(),
+        Matchers.<WindowedValue<Long>>emptyIterable());
+  }
+
+  @Test
   public void evaluatorClosesReaderAfterOutputCount() throws Exception {
     ContiguousSet<Long> elems = ContiguousSet.create(
         Range.closed(0L, 20L * UnboundedReadEvaluatorFactory.MAX_READER_REUSE_COUNT),
@@ -251,6 +284,7 @@ public class UnboundedReadEvaluatorFactoryTest {
     static int readerAdvancedCount;
     private final Coder<T> coder;
     private final List<T> elems;
+    private boolean dedupes = false;
 
     public TestUnboundedSource(Coder<T> coder, T... elems) {
       readerAdvancedCount = 0;
@@ -278,6 +312,11 @@ public class UnboundedReadEvaluatorFactoryTest {
     }
 
     @Override
+    public boolean requiresDeduping() {
+      return dedupes;
+    }
+
+    @Override
     public void validate() {}
 
     @Override
@@ -332,7 +371,16 @@ public class UnboundedReadEvaluatorFactoryTest {
 
       @Override
       public Instant getCurrentTimestamp() throws NoSuchElementException {
-        return Instant.now();
+        return new Instant(index);
+      }
+
+      @Override
+      public byte[] getCurrentRecordId() {
+        try {
+          return CoderUtils.encodeToByteArray(coder, getCurrent());
+        } catch (CoderException e) {
+          throw new RuntimeException(e);
+        }
       }
 
       @Override


Mime
View raw message