You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@apex.apache.org by bhupeshchawda <gi...@git.apache.org> on 2016/07/14 13:26:15 UTC
[GitHub] apex-malhar pull request #343: [Review Only] APEXMALHAR-1701: Deduper using ...
GitHub user bhupeshchawda opened a pull request:
https://github.com/apache/apex-malhar/pull/343
[Review Only] APEXMALHAR-1701: Deduper using Windowed Operator - Initial Commit. Ad\u2026
\u2026ded Tests
Added an implementation based on Windowed Operator
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/bhupeshchawda/apex-malhar APEXMALHAR-1701-windowed-deduper
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/apex-malhar/pull/343.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #343
----
commit bef469e7a3b2054915185060d19c59f1dc2a94cb
Author: bhupesh <bh...@gmail.com>
Date: 2016-07-13T08:22:45Z
APEXMALHAR-1701: Deduper using Windowed Operator - Initial Commit. Added Tests
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] apex-malhar pull request #343: [Review Only] APEXMALHAR-1701: Deduper using ...
Posted by bhupeshchawda <gi...@git.apache.org>.
Github user bhupeshchawda commented on a diff in the pull request:
https://github.com/apache/apex-malhar/pull/343#discussion_r70945247
--- Diff: library/src/main/java/org/apache/apex/malhar/lib/dedup/AbstractWindowedDeduper.java ---
@@ -0,0 +1,97 @@
+/**
+ * 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.apex.malhar.lib.dedup;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.apex.malhar.lib.window.Window;
+import org.apache.apex.malhar.lib.window.impl.WindowedOperatorImpl;
+
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+/**
+ * Abstract implementation for the Windowed Dedup operator
+ *
+ * @param <T>
+ */
+public abstract class AbstractWindowedDeduper<T> extends WindowedOperatorImpl<T, List<T>, T>
+{
+
+ /**
+ * The output port on which unique events are emitted.
+ */
+ public final transient DefaultOutputPort<T> unique = new DefaultOutputPort<T>();
+
+ /**
+ * The output port on which duplicate events are emitted.
+ */
+ @OutputPortFieldAnnotation(optional = true)
+ public final transient DefaultOutputPort<T> duplicates = new DefaultOutputPort<T>();
+
+ /**
+ * The output port on which expired events are emitted.
+ */
+ @OutputPortFieldAnnotation(optional = true)
+ public final transient DefaultOutputPort<T> expired = new DefaultOutputPort<T>();
+
+ /**
+ * Returns the key of the tuple based on which the tuples are de-duplicated
+ */
+ protected abstract Object getKey(T tuple);
+
+ /**
+ * Insead of dropping, emit on expired port
+ */
+ @Override
+ public void dropTuple(Tuple<T> input)
+ {
+ expired.emit(input.getValue());
+ }
+
+ /**
+ * Performs the de-duplication
+ */
+ @Override
+ public void accumulateTuple(Tuple.WindowedTuple<T> tuple)
+ {
+ for (Window window : tuple.getWindows()) { // There will be exactly one window
+ // process each window
+ List<T> accum = dataStorage.get(window);
+ if (accum == null) {
+ // New window?
+ accum = new ArrayList<T>();
--- End diff --
agreed. Will change this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] apex-malhar pull request #343: [Review Only] APEXMALHAR-1701: Deduper using ...
Posted by bhupeshchawda <gi...@git.apache.org>.
Github user bhupeshchawda commented on a diff in the pull request:
https://github.com/apache/apex-malhar/pull/343#discussion_r70945182
--- Diff: library/src/test/java/org/apache/apex/malhar/lib/dedup/WindowedDeduperTest.java ---
@@ -0,0 +1,160 @@
+/**
+ * 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.apex.malhar.lib.dedup;
+
+import java.util.Date;
+import java.util.List;
+import java.util.Random;
+
+import org.joda.time.Duration;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.apex.malhar.lib.window.Accumulation;
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.apex.malhar.lib.window.Tuple.TimestampedTuple;
+import org.apache.apex.malhar.lib.window.WindowOption;
+import org.apache.apex.malhar.lib.window.WindowState;
+import org.apache.apex.malhar.lib.window.impl.InMemoryWindowedStorage;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.helper.OperatorContextTestHelper;
+import com.datatorrent.lib.testbench.CollectorTestSink;
+import com.datatorrent.lib.util.TestUtils;
+import com.datatorrent.stram.engine.WindowGenerator;
+
+public class WindowedDeduperTest
+{
+ private static final int OPERATOR_ID = 0;
+ private static AbstractWindowedDeduper<TestPojo> dedup;
+ private static final int WINDOW_SPAN = 10;
+ private static final int FIXED_WATERMARK = 30;
+
+ /**
+ * Windowed Deduper Test Implementation
+ * Tuples of type {@link TestPojo}
+ */
+ public static class WindowedDeduperImpl extends AbstractWindowedDeduper<TestPojo>
+ {
+ @Override
+ protected Object getKey(TestPojo tuple)
+ {
+ return tuple.getKey();
+ }
+ }
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ @BeforeClass
+ public static void setup()
+ {
+ dedup = new WindowedDeduperImpl();
+ // Define accumulation
+ Accumulation dedupAccum = new DedupAccumulation<TestPojo>();
+ dedup.setAccumulation(dedupAccum);
+ // Set storage to InMemory implementation
+ dedup.setDataStorage(new InMemoryWindowedStorage<List<TestPojo>>());
+ // Set retraction storage. Not used, but is required by the abstract operator.
+ dedup.setRetractionStorage(new InMemoryWindowedStorage<TestPojo>());
--- End diff --
Yes. It throws a NPE. Changes are needed in the abstract operator.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] apex-malhar pull request #343: [Review Only] APEXMALHAR-1701: Deduper using ...
Posted by davidyan74 <gi...@git.apache.org>.
Github user davidyan74 commented on a diff in the pull request:
https://github.com/apache/apex-malhar/pull/343#discussion_r70899585
--- Diff: library/src/test/java/org/apache/apex/malhar/lib/dedup/WindowedDeduperTest.java ---
@@ -0,0 +1,160 @@
+/**
+ * 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.apex.malhar.lib.dedup;
+
+import java.util.Date;
+import java.util.List;
+import java.util.Random;
+
+import org.joda.time.Duration;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.apex.malhar.lib.window.Accumulation;
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.apex.malhar.lib.window.Tuple.TimestampedTuple;
+import org.apache.apex.malhar.lib.window.WindowOption;
+import org.apache.apex.malhar.lib.window.WindowState;
+import org.apache.apex.malhar.lib.window.impl.InMemoryWindowedStorage;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.helper.OperatorContextTestHelper;
+import com.datatorrent.lib.testbench.CollectorTestSink;
+import com.datatorrent.lib.util.TestUtils;
+import com.datatorrent.stram.engine.WindowGenerator;
+
+public class WindowedDeduperTest
+{
+ private static final int OPERATOR_ID = 0;
+ private static AbstractWindowedDeduper<TestPojo> dedup;
+ private static final int WINDOW_SPAN = 10;
+ private static final int FIXED_WATERMARK = 30;
+
+ /**
+ * Windowed Deduper Test Implementation
+ * Tuples of type {@link TestPojo}
+ */
+ public static class WindowedDeduperImpl extends AbstractWindowedDeduper<TestPojo>
+ {
+ @Override
+ protected Object getKey(TestPojo tuple)
+ {
+ return tuple.getKey();
+ }
+ }
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ @BeforeClass
+ public static void setup()
+ {
+ dedup = new WindowedDeduperImpl();
+ // Define accumulation
+ Accumulation dedupAccum = new DedupAccumulation<TestPojo>();
+ dedup.setAccumulation(dedupAccum);
+ // Set storage to InMemory implementation
+ dedup.setDataStorage(new InMemoryWindowedStorage<List<TestPojo>>());
+ // Set retraction storage. Not used, but is required by the abstract operator.
+ dedup.setRetractionStorage(new InMemoryWindowedStorage<TestPojo>());
--- End diff --
Should not need this if you're not doing triggers at all here. Does it give you an error?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] apex-malhar pull request #343: [Review Only] APEXMALHAR-1701: Deduper using ...
Posted by davidyan74 <gi...@git.apache.org>.
Github user davidyan74 commented on a diff in the pull request:
https://github.com/apache/apex-malhar/pull/343#discussion_r70899817
--- Diff: library/src/main/java/org/apache/apex/malhar/lib/dedup/AbstractWindowedDeduper.java ---
@@ -0,0 +1,97 @@
+/**
+ * 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.apex.malhar.lib.dedup;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.apex.malhar.lib.window.Window;
+import org.apache.apex.malhar.lib.window.impl.WindowedOperatorImpl;
+
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+/**
+ * Abstract implementation for the Windowed Dedup operator
+ *
+ * @param <T>
+ */
+public abstract class AbstractWindowedDeduper<T> extends WindowedOperatorImpl<T, List<T>, T>
+{
+
+ /**
+ * The output port on which unique events are emitted.
+ */
+ public final transient DefaultOutputPort<T> unique = new DefaultOutputPort<T>();
+
+ /**
+ * The output port on which duplicate events are emitted.
+ */
+ @OutputPortFieldAnnotation(optional = true)
+ public final transient DefaultOutputPort<T> duplicates = new DefaultOutputPort<T>();
+
+ /**
+ * The output port on which expired events are emitted.
+ */
+ @OutputPortFieldAnnotation(optional = true)
+ public final transient DefaultOutputPort<T> expired = new DefaultOutputPort<T>();
+
+ /**
+ * Returns the key of the tuple based on which the tuples are de-duplicated
+ */
+ protected abstract Object getKey(T tuple);
+
+ /**
+ * Insead of dropping, emit on expired port
+ */
+ @Override
+ public void dropTuple(Tuple<T> input)
+ {
+ expired.emit(input.getValue());
+ }
+
+ /**
+ * Performs the de-duplication
+ */
+ @Override
+ public void accumulateTuple(Tuple.WindowedTuple<T> tuple)
+ {
+ for (Window window : tuple.getWindows()) { // There will be exactly one window
+ // process each window
+ List<T> accum = dataStorage.get(window);
+ if (accum == null) {
+ // New window?
+ accum = new ArrayList<T>();
--- End diff --
I think it's better to call defaultAccumulatedValue() here and have defaultAccumulatedValue return a new list.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] apex-malhar pull request #343: [Review Only] APEXMALHAR-1701: Deduper using ...
Posted by bhupeshchawda <gi...@git.apache.org>.
Github user bhupeshchawda closed the pull request at:
https://github.com/apache/apex-malhar/pull/343
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] apex-malhar pull request #343: [Review Only] APEXMALHAR-1701: Deduper using ...
Posted by davidyan74 <gi...@git.apache.org>.
Github user davidyan74 commented on a diff in the pull request:
https://github.com/apache/apex-malhar/pull/343#discussion_r71046708
--- Diff: library/src/test/java/org/apache/apex/malhar/lib/dedup/WindowedDeduperTest.java ---
@@ -0,0 +1,160 @@
+/**
+ * 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.apex.malhar.lib.dedup;
+
+import java.util.Date;
+import java.util.List;
+import java.util.Random;
+
+import org.joda.time.Duration;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.apex.malhar.lib.window.Accumulation;
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.apex.malhar.lib.window.Tuple.TimestampedTuple;
+import org.apache.apex.malhar.lib.window.WindowOption;
+import org.apache.apex.malhar.lib.window.WindowState;
+import org.apache.apex.malhar.lib.window.impl.InMemoryWindowedStorage;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.helper.OperatorContextTestHelper;
+import com.datatorrent.lib.testbench.CollectorTestSink;
+import com.datatorrent.lib.util.TestUtils;
+import com.datatorrent.stram.engine.WindowGenerator;
+
+public class WindowedDeduperTest
+{
+ private static final int OPERATOR_ID = 0;
+ private static AbstractWindowedDeduper<TestPojo> dedup;
+ private static final int WINDOW_SPAN = 10;
+ private static final int FIXED_WATERMARK = 30;
+
+ /**
+ * Windowed Deduper Test Implementation
+ * Tuples of type {@link TestPojo}
+ */
+ public static class WindowedDeduperImpl extends AbstractWindowedDeduper<TestPojo>
+ {
+ @Override
+ protected Object getKey(TestPojo tuple)
+ {
+ return tuple.getKey();
+ }
+ }
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ @BeforeClass
+ public static void setup()
+ {
+ dedup = new WindowedDeduperImpl();
+ // Define accumulation
+ Accumulation dedupAccum = new DedupAccumulation<TestPojo>();
+ dedup.setAccumulation(dedupAccum);
+ // Set storage to InMemory implementation
+ dedup.setDataStorage(new InMemoryWindowedStorage<List<TestPojo>>());
+ // Set retraction storage. Not used, but is required by the abstract operator.
+ dedup.setRetractionStorage(new InMemoryWindowedStorage<TestPojo>());
--- End diff --
I see where the problem is. I have a fix here in my newest review-only PR: #345
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---