You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by dh...@apache.org on 2016/06/22 16:57:27 UTC
[1/2] incubator-beam git commit: Deduplicate Unbounded Reads
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 <tg...@google.com>
Authored: Wed Jun 15 14:41:09 2016 -0700
Committer: Dan Halperin <dh...@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
[2/2] incubator-beam git commit: Closes #473
Posted by dh...@apache.org.
Closes #473
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/c83e5c48
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/c83e5c48
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/c83e5c48
Branch: refs/heads/master
Commit: c83e5c48f002003f60e06a3dd41c83aa2a702a9a
Parents: 3bc0fe6 569228e
Author: Dan Halperin <dh...@google.com>
Authored: Wed Jun 22 09:57:24 2016 -0700
Committer: Dan Halperin <dh...@google.com>
Committed: Wed Jun 22 09:57:24 2016 -0700
----------------------------------------------------------------------
.../direct/UnboundedReadDeduplicator.java | 102 ++++++++++++++
.../direct/UnboundedReadEvaluatorFactory.java | 17 ++-
.../direct/UnboundedReadDeduplicatorTest.java | 134 +++++++++++++++++++
.../UnboundedReadEvaluatorFactoryTest.java | 50 ++++++-
4 files changed, 299 insertions(+), 4 deletions(-)
----------------------------------------------------------------------