You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@edgent.apache.org by wc...@apache.org on 2016/04/06 02:04:08 UTC
[1/3] incubator-quarks git commit: Added join and joinLast
functionality.
Repository: incubator-quarks
Updated Branches:
refs/heads/master 9dd8e314f -> f58e8f596
Added join and joinLast functionality.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quarks/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quarks/commit/7aa29efb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quarks/tree/7aa29efb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quarks/diff/7aa29efb
Branch: refs/heads/master
Commit: 7aa29efbb55347aa4ea9e24b6b2110af49e86d08
Parents: 9dd8e31
Author: wcmarsha <wc...@us.ibm.com>
Authored: Thu Mar 31 10:11:20 2016 -0700
Committer: wcmarsha <wc...@us.ibm.com>
Committed: Tue Apr 5 10:08:50 2016 -0700
----------------------------------------------------------------------
.../src/main/java/quarks/topology/TStream.java | 8 ++
.../src/main/java/quarks/window/Window.java | 7 ++
.../src/main/java/quarks/window/WindowImpl.java | 7 +-
.../topology/spi/graph/ConnectorStream.java | 108 +++++++++++++++++++
.../quarks/topology/spi/graph/TWindowImpl.java | 7 ++
.../topology/spi/graph/TWindowTimeImpl.java | 28 +++++
6 files changed, 164 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/7aa29efb/api/topology/src/main/java/quarks/topology/TStream.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/quarks/topology/TStream.java b/api/topology/src/main/java/quarks/topology/TStream.java
index 8d23e6b..275710b 100644
--- a/api/topology/src/main/java/quarks/topology/TStream.java
+++ b/api/topology/src/main/java/quarks/topology/TStream.java
@@ -18,6 +18,7 @@ under the License.
*/
package quarks.topology;
+import quarks.function.BiFunction;
import quarks.function.Consumer;
import quarks.function.Function;
import quarks.function.Predicate;
@@ -439,4 +440,11 @@ public interface TStream<T> extends TopologyElement {
* @return set of tags
*/
Set<String> getTags();
+
+ <J, U, K> TStream<J> join(Function<T, K> keyer, TWindow<U, K> window, BiFunction<T, List<U>, J> joiner);
+
+ <J, U, K> TStream<J> joinLast(Function<T, K> keyer, TStream<U> lastStream, Function<U, K> lastStreamKeyer, BiFunction<T, U, J> joiner);
+
+ <J, U> TStream<J> joinLast(TStream<U> lastStream, BiFunction<T, U, J> joiner);
+
}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/7aa29efb/api/window/src/main/java/quarks/window/Window.java
----------------------------------------------------------------------
diff --git a/api/window/src/main/java/quarks/window/Window.java b/api/window/src/main/java/quarks/window/Window.java
index 02cb40f..f78a744 100644
--- a/api/window/src/main/java/quarks/window/Window.java
+++ b/api/window/src/main/java/quarks/window/Window.java
@@ -19,6 +19,7 @@ under the License.
package quarks.window;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import quarks.function.BiConsumer;
@@ -157,4 +158,10 @@ public interface Window<T, K, L extends List<T>>{
*/
Function<T, K> getKeyFunction();
+ /**
+ * Retrieves the partitions in the window.
+ * @return A map of the window's keys and partitions.
+ */
+ Map<K, Partition<T, K, L>> getPartitions();
+
}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/7aa29efb/api/window/src/main/java/quarks/window/WindowImpl.java
----------------------------------------------------------------------
diff --git a/api/window/src/main/java/quarks/window/WindowImpl.java b/api/window/src/main/java/quarks/window/WindowImpl.java
index 7ecb79f..a4e308b 100644
--- a/api/window/src/main/java/quarks/window/WindowImpl.java
+++ b/api/window/src/main/java/quarks/window/WindowImpl.java
@@ -42,7 +42,7 @@ class WindowImpl<T, K, L extends List<T>> implements Window<T, K, L> {
protected Supplier<L> listSupplier;
protected Function<T, K> keyFunction;
- protected Map<K, Partition<T, K, L> > partitions = new HashMap<K, Partition<T, K, L> >();
+ private Map<K, Partition<T, K, L> > partitions = new HashMap<K, Partition<T, K, L> >();
WindowImpl(BiFunction<Partition<T, K, L>, T, Boolean> insertionPolicy, BiConsumer<Partition<T, K, L>, T> contentsPolicy,
@@ -119,4 +119,9 @@ class WindowImpl<T, K, L extends List<T>> implements Window<T, K, L> {
return this.ses;
}
+ @Override
+ public Map<K, Partition<T, K, L>> getPartitions() {
+ return partitions;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/7aa29efb/spi/topology/src/main/java/quarks/topology/spi/graph/ConnectorStream.java
----------------------------------------------------------------------
diff --git a/spi/topology/src/main/java/quarks/topology/spi/graph/ConnectorStream.java b/spi/topology/src/main/java/quarks/topology/spi/graph/ConnectorStream.java
index 42385fb..c75693c 100644
--- a/spi/topology/src/main/java/quarks/topology/spi/graph/ConnectorStream.java
+++ b/spi/topology/src/main/java/quarks/topology/spi/graph/ConnectorStream.java
@@ -19,14 +19,21 @@ under the License.
package quarks.topology.spi.graph;
import static quarks.function.Functions.synchronizedFunction;
+import static quarks.window.Policies.alwaysInsert;
+import static quarks.window.Policies.evictOlderWithProcess;
+import static quarks.window.Policies.insertionTimeList;
+import static quarks.window.Policies.processOnInsert;
+import static quarks.window.Policies.scheduleEvictIfEmpty;
import java.util.ArrayList;
import java.util.EnumMap;
import java.util.HashSet;
+import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import quarks.function.BiFunction;
import quarks.function.Consumer;
import quarks.function.Function;
import quarks.function.Functions;
@@ -43,11 +50,16 @@ import quarks.oplet.functional.Filter;
import quarks.oplet.functional.FlatMap;
import quarks.oplet.functional.Map;
import quarks.oplet.functional.Peek;
+import quarks.oplet.window.Aggregate;
import quarks.topology.TSink;
import quarks.topology.TStream;
import quarks.topology.TWindow;
import quarks.topology.Topology;
import quarks.topology.spi.AbstractTStream;
+import quarks.window.InsertionTimeList;
+import quarks.window.Partition;
+import quarks.window.Window;
+import quarks.window.Windows;
/**
* A stream that directly adds oplets to the graph.
@@ -163,6 +175,100 @@ public class ConnectorStream<G extends Topology, T> extends AbstractTStream<G, T
}
@Override
+ public <J, U, K> TStream<J> join(Function<T, K> keyer,
+ TWindow<U, K> twindow, BiFunction<T, List<U>, J> joiner) {
+
+ if(twindow instanceof TWindowImpl){
+ TStream<U> lastStream = twindow.feeder();
+ BiFunction<List<U>,K, Object> processor = Functions.synchronizedBiFunction((list, key) -> null);
+ Window<U, K, LinkedList<U>> window = Windows.lastNProcessOnInsert(((TWindowImpl<U, K>)twindow).getSize(), twindow.getKeyFunction());
+ Aggregate<U,Object,K> op = new Aggregate<U,Object,K>(window, processor);
+ lastStream.pipe(op);
+ return this.map((tuple) -> {
+ Partition<U, K, ? extends List<U>> part = window.getPartitions().get(keyer.apply(tuple));
+ if(part == null)
+ return null;
+ J ret;
+ synchronized (part) {
+ List<U> last = part.getContents();
+ ret = joiner.apply(tuple, last);
+ }
+ return ret;
+ });
+ }
+
+ else if (twindow instanceof TWindowTimeImpl){
+ TStream<U> lastStream = twindow.feeder();
+ BiFunction<List<U>,K, Object> processor = Functions.synchronizedBiFunction((list, key) -> null);
+ long time = ((TWindowTimeImpl<U, K>)(twindow)).getTime();
+ TimeUnit unit = ((TWindowTimeImpl<U, K>)(twindow)).getUnit();
+ Window<U, K, InsertionTimeList<U>> window =
+ Windows.window(
+ alwaysInsert(),
+ scheduleEvictIfEmpty(time, unit),
+ evictOlderWithProcess(time, unit),
+ processOnInsert(),
+ twindow.getKeyFunction(),
+ insertionTimeList());
+ Aggregate<U,Object,K> op = new Aggregate<U,Object,K>(window, processor);
+ lastStream.pipe(op);
+ return this.map((tuple) -> {
+ Partition<U, K, ? extends List<U>> part = window.getPartitions().get(keyer.apply(tuple));
+ if(part == null)
+ return null;
+ J ret;
+ synchronized (part) {
+ List<U> last = part.getContents();
+ ret = joiner.apply(tuple, last);
+ }
+ return ret;
+ });
+ }
+ else{
+ throw new IllegalStateException("Unsupported window format");
+ }
+ }
+
+ @Override
+ public <J, U, K> TStream<J> joinLast(Function<T, K> keyer,
+ TStream<U> lastStream, Function<U, K> lastStreamKeyer, BiFunction<T, U, J> joiner) {
+ BiFunction<List<U>,K, Object> processor = Functions.synchronizedBiFunction((list, key) -> null);
+ Window<U, K, LinkedList<U>> window = Windows.lastNProcessOnInsert(1, lastStreamKeyer);
+ Aggregate<U,Object,K> op = new Aggregate<U,Object,K>(window, processor);
+ lastStream.pipe(op);
+ return this.map((tuple) -> {
+ Partition<U, K, ? extends List<U>> part = window.getPartitions().get(keyer.apply(tuple));
+ if(part == null)
+ return null;
+ J ret;
+ synchronized (part) {
+ U last = part.getContents().get(0);
+ ret = joiner.apply(tuple, last);
+ }
+ return ret;
+ });
+ }
+
+ @Override
+ public <J, U> TStream<J> joinLast(TStream<U> lastStream, BiFunction<T, U, J> joiner) {
+ BiFunction<List<U>,Integer, Object> processor = Functions.synchronizedBiFunction((list, key) -> null);
+ Window<U, Integer, LinkedList<U>> window = Windows.lastNProcessOnInsert(1, tuple -> 0);
+ Aggregate<U,Object,Integer> op = new Aggregate<U,Object,Integer>(window, processor);
+ lastStream.pipe(op);
+ return this.map((tuple) -> {
+ Partition<U, Integer, ? extends List<U>> part = window.getPartitions().get(0);
+ if(part == null)
+ return null;
+ J ret;
+ synchronized (part) {
+ U last = part.getContents().get(0);
+ ret = joiner.apply(tuple, last);
+ }
+ return ret;
+ });
+ }
+
+ @Override
public TStream<T> union(Set<TStream<T>> others) {
if (others.isEmpty())
return this;
@@ -200,4 +306,6 @@ public class ConnectorStream<G extends Topology, T> extends AbstractTStream<G, T
return connector.getTags();
}
+
+
}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/7aa29efb/spi/topology/src/main/java/quarks/topology/spi/graph/TWindowImpl.java
----------------------------------------------------------------------
diff --git a/spi/topology/src/main/java/quarks/topology/spi/graph/TWindowImpl.java b/spi/topology/src/main/java/quarks/topology/spi/graph/TWindowImpl.java
index 5a8e5c8..cd9fee2 100644
--- a/spi/topology/src/main/java/quarks/topology/spi/graph/TWindowImpl.java
+++ b/spi/topology/src/main/java/quarks/topology/spi/graph/TWindowImpl.java
@@ -64,4 +64,11 @@ public class TWindowImpl<T, K> extends AbstractTWindow<T, K> {
Aggregate<T,U,K> op = new Aggregate<T,U,K>(window, batcher);
return feeder().pipe(op);
}
+
+ /**
+ * @return the size
+ */
+ public int getSize() {
+ return size;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/7aa29efb/spi/topology/src/main/java/quarks/topology/spi/graph/TWindowTimeImpl.java
----------------------------------------------------------------------
diff --git a/spi/topology/src/main/java/quarks/topology/spi/graph/TWindowTimeImpl.java b/spi/topology/src/main/java/quarks/topology/spi/graph/TWindowTimeImpl.java
index 8565ace..6194a05 100644
--- a/spi/topology/src/main/java/quarks/topology/spi/graph/TWindowTimeImpl.java
+++ b/spi/topology/src/main/java/quarks/topology/spi/graph/TWindowTimeImpl.java
@@ -94,5 +94,33 @@ public class TWindowTimeImpl<T, K> extends AbstractTWindow<T, K> {
Aggregate<T,U,K> op = new Aggregate<T,U,K>(window, batcher);
return feeder().pipe(op);
+ }
+
+ /**
+ * @return the time
+ */
+ public long getTime() {
+ return time;
+ }
+
+ /**
+ * @param time the time to set
+ */
+ public void setTime(long time) {
+ this.time = time;
+ }
+
+ /**
+ * @return the unit
+ */
+ public TimeUnit getUnit() {
+ return unit;
+ }
+
+ /**
+ * @param unit the unit to set
+ */
+ public void setUnit(TimeUnit unit) {
+ this.unit = unit;
}
}
[3/3] incubator-quarks git commit: Updated doc. Minor refactoring.
Posted by wc...@apache.org.
Updated doc. Minor refactoring.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quarks/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quarks/commit/f58e8f59
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quarks/tree/f58e8f59
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quarks/diff/f58e8f59
Branch: refs/heads/master
Commit: f58e8f59602fb4e21ad174bdabff9dce374d59d5
Parents: 678dc18
Author: wcmarsha <wc...@us.ibm.com>
Authored: Mon Apr 4 23:30:07 2016 -0700
Committer: wcmarsha <wc...@us.ibm.com>
Committed: Tue Apr 5 10:08:54 2016 -0700
----------------------------------------------------------------------
.../src/main/java/quarks/topology/TStream.java | 41 +++++++++-
.../src/main/java/quarks/window/Partition.java | 11 ++-
.../src/main/java/quarks/window/Window.java | 12 ++-
.../topology/spi/graph/ConnectorStream.java | 84 ++++++++------------
.../topology/spi/graph/TWindowTimeImpl.java | 13 ---
5 files changed, 92 insertions(+), 69 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/f58e8f59/api/topology/src/main/java/quarks/topology/TStream.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/quarks/topology/TStream.java b/api/topology/src/main/java/quarks/topology/TStream.java
index 275710b..c11c0d7 100644
--- a/api/topology/src/main/java/quarks/topology/TStream.java
+++ b/api/topology/src/main/java/quarks/topology/TStream.java
@@ -441,10 +441,47 @@ public interface TStream<T> extends TopologyElement {
*/
Set<String> getTags();
+ /**
+ * Join this stream with a partitioned window of type {@code U} with key type {@code K}.
+ * For each tuple on this stream, it is joined with the contents of {@code window}
+ * for the key {@code keyer.apply(tuple)}. Each tuple is
+ * passed into {@code joiner} and the return value is submitted to the
+ * returned stream. If call returns null then no tuple is submitted.
+ *
+ * @param keyer Key function for this stream to match the window's key.
+ * @param window Keyed window to join this stream with.
+ * @param joiner Join function.
+ * @return A stream that is the results of joining this stream with
+ * {@code window}.
+ */
<J, U, K> TStream<J> join(Function<T, K> keyer, TWindow<U, K> window, BiFunction<T, List<U>, J> joiner);
+ /**
+ * Join this stream with the last tuple seen on a stream of type {@code U}
+ * with partitioning.
+ * For each tuple on this
+ * stream, it is joined with the last tuple seen on {@code lastStream}
+ * with a matching key (of type {@code K}).
+ * <BR>
+ * Each tuple {@code t} on this stream will match the last tuple
+ * {@code u} on {@code lastStream} if
+ * {@code keyer.apply(t).equals(lastStreamKeyer.apply(u))}
+ * is true.
+ * <BR>
+ * The assumption is made that
+ * the key classes correctly implement the contract for {@code equals} and
+ * {@code hashCode()}.
+ * <P>Each tuple is
+ * passed into {@code joiner} and the return value is submitted to the
+ * returned stream. If call returns null then no tuple is submitted.
+ * </P>
+ * @param keyer Key function for this stream
+ * @param lastStream Stream to join with.
+ * @param lastStreamKeyer Key function for {@code lastStream}
+ * @param joiner Join function.
+ * @return A stream that is the results of joining this stream with
+ * {@code lastStream}.
+ */
<J, U, K> TStream<J> joinLast(Function<T, K> keyer, TStream<U> lastStream, Function<U, K> lastStreamKeyer, BiFunction<T, U, J> joiner);
-
- <J, U> TStream<J> joinLast(TStream<U> lastStream, BiFunction<T, U, J> joiner);
}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/f58e8f59/api/window/src/main/java/quarks/window/Partition.java
----------------------------------------------------------------------
diff --git a/api/window/src/main/java/quarks/window/Partition.java b/api/window/src/main/java/quarks/window/Partition.java
index dff651d..4b80c46 100644
--- a/api/window/src/main/java/quarks/window/Partition.java
+++ b/api/window/src/main/java/quarks/window/Partition.java
@@ -22,7 +22,16 @@ import java.io.Serializable;
import java.util.List;
/**
- * A partition within a {@code Window}.
+ * A partition within a {@code Window}. The contents of the list
+ * returned by {@code getContents} is stable when synchronizing
+ * on the partition object. For example:
+ *
+ * <pre><code>
+ * Partition<Integer, Integer, ArrayList<Integer>> part = ...;
+ * synchronized(part){
+ * List<Integer> = part.getContents();
+ * // stable operation on contents of partition
+ * }
*
* @param <T> Type of tuples in the partition.
* @param <K> Type of the partition's key.
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/f58e8f59/api/window/src/main/java/quarks/window/Window.java
----------------------------------------------------------------------
diff --git a/api/window/src/main/java/quarks/window/Window.java b/api/window/src/main/java/quarks/window/Window.java
index f78a744..a9ac191 100644
--- a/api/window/src/main/java/quarks/window/Window.java
+++ b/api/window/src/main/java/quarks/window/Window.java
@@ -159,7 +159,17 @@ public interface Window<T, K, L extends List<T>>{
Function<T, K> getKeyFunction();
/**
- * Retrieves the partitions in the window.
+ * Retrieves the partitions in the window. The map of partitions
+ * is stable when synchronizing on the intrinsic lock of the map,
+ * for example:
+ * <br>
+ * <pre><code>
+ * Map<K, Partitions<U, K, ?>> partitions = window.getPartitions();
+ * synchronized(partitions){
+ * // operations with partition
+ * }
+ * </code></pre>
+ *
* @return A map of the window's keys and partitions.
*/
Map<K, Partition<T, K, L>> getPartitions();
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/f58e8f59/spi/topology/src/main/java/quarks/topology/spi/graph/ConnectorStream.java
----------------------------------------------------------------------
diff --git a/spi/topology/src/main/java/quarks/topology/spi/graph/ConnectorStream.java b/spi/topology/src/main/java/quarks/topology/spi/graph/ConnectorStream.java
index c75693c..36c1f5d 100644
--- a/spi/topology/src/main/java/quarks/topology/spi/graph/ConnectorStream.java
+++ b/spi/topology/src/main/java/quarks/topology/spi/graph/ConnectorStream.java
@@ -22,7 +22,6 @@ import static quarks.function.Functions.synchronizedFunction;
import static quarks.window.Policies.alwaysInsert;
import static quarks.window.Policies.evictOlderWithProcess;
import static quarks.window.Policies.insertionTimeList;
-import static quarks.window.Policies.processOnInsert;
import static quarks.window.Policies.scheduleEvictIfEmpty;
import java.util.ArrayList;
@@ -56,8 +55,8 @@ import quarks.topology.TStream;
import quarks.topology.TWindow;
import quarks.topology.Topology;
import quarks.topology.spi.AbstractTStream;
-import quarks.window.InsertionTimeList;
import quarks.window.Partition;
+import quarks.window.Policies;
import quarks.window.Window;
import quarks.window.Windows;
@@ -178,85 +177,66 @@ public class ConnectorStream<G extends Topology, T> extends AbstractTStream<G, T
public <J, U, K> TStream<J> join(Function<T, K> keyer,
TWindow<U, K> twindow, BiFunction<T, List<U>, J> joiner) {
- if(twindow instanceof TWindowImpl){
- TStream<U> lastStream = twindow.feeder();
- BiFunction<List<U>,K, Object> processor = Functions.synchronizedBiFunction((list, key) -> null);
- Window<U, K, LinkedList<U>> window = Windows.lastNProcessOnInsert(((TWindowImpl<U, K>)twindow).getSize(), twindow.getKeyFunction());
- Aggregate<U,Object,K> op = new Aggregate<U,Object,K>(window, processor);
- lastStream.pipe(op);
- return this.map((tuple) -> {
- Partition<U, K, ? extends List<U>> part = window.getPartitions().get(keyer.apply(tuple));
- if(part == null)
- return null;
- J ret;
- synchronized (part) {
- List<U> last = part.getContents();
- ret = joiner.apply(tuple, last);
- }
- return ret;
- });
+ TStream<U> lastStream = twindow.feeder();
+ BiFunction<List<U>,K, Object> processor = Functions.synchronizedBiFunction((list, key) -> null);
+ Window<U, K, ?> window;
+ if(twindow instanceof TWindowImpl){
+ window = Windows.lastNProcessOnInsert(((TWindowImpl<U, K>)twindow).getSize(), twindow.getKeyFunction());
+
}
else if (twindow instanceof TWindowTimeImpl){
- TStream<U> lastStream = twindow.feeder();
- BiFunction<List<U>,K, Object> processor = Functions.synchronizedBiFunction((list, key) -> null);
long time = ((TWindowTimeImpl<U, K>)(twindow)).getTime();
TimeUnit unit = ((TWindowTimeImpl<U, K>)(twindow)).getUnit();
- Window<U, K, InsertionTimeList<U>> window =
- Windows.window(
+ window = Windows.window(
alwaysInsert(),
scheduleEvictIfEmpty(time, unit),
evictOlderWithProcess(time, unit),
- processOnInsert(),
+ Policies.doNothing(),
twindow.getKeyFunction(),
insertionTimeList());
- Aggregate<U,Object,K> op = new Aggregate<U,Object,K>(window, processor);
- lastStream.pipe(op);
- return this.map((tuple) -> {
- Partition<U, K, ? extends List<U>> part = window.getPartitions().get(keyer.apply(tuple));
- if(part == null)
- return null;
- J ret;
- synchronized (part) {
- List<U> last = part.getContents();
- ret = joiner.apply(tuple, last);
- }
- return ret;
- });
}
else{
throw new IllegalStateException("Unsupported window format");
}
- }
-
- @Override
- public <J, U, K> TStream<J> joinLast(Function<T, K> keyer,
- TStream<U> lastStream, Function<U, K> lastStreamKeyer, BiFunction<T, U, J> joiner) {
- BiFunction<List<U>,K, Object> processor = Functions.synchronizedBiFunction((list, key) -> null);
- Window<U, K, LinkedList<U>> window = Windows.lastNProcessOnInsert(1, lastStreamKeyer);
+
+ // To perform a join, the runtime needs to maintain a windowImpl based on
+ // the tuples from the twindow.feeder TStream. To do this, it's
+ // necessary to create an Aggregate oplet and insert it into the
+ // graph with lastStream.pipe.
Aggregate<U,Object,K> op = new Aggregate<U,Object,K>(window, processor);
lastStream.pipe(op);
+
return this.map((tuple) -> {
- Partition<U, K, ? extends List<U>> part = window.getPartitions().get(keyer.apply(tuple));
+ // The window object can be referenced via closure, and the corresponding
+ // partition can be retrieved based on the keyer. This way, we avoid
+ // needing to create an additional oplet type with multiple input ports.
+
+ java.util.Map<K, ?> partitions = window.getPartitions();
+ Partition<U, K, ? extends List<U>> part;
+ synchronized(partitions){
+ part = window.getPartitions().get(keyer.apply(tuple));
+ }
if(part == null)
return null;
J ret;
synchronized (part) {
- U last = part.getContents().get(0);
+ List<U> last = part.getContents();
ret = joiner.apply(tuple, last);
}
return ret;
});
}
-
+
@Override
- public <J, U> TStream<J> joinLast(TStream<U> lastStream, BiFunction<T, U, J> joiner) {
- BiFunction<List<U>,Integer, Object> processor = Functions.synchronizedBiFunction((list, key) -> null);
- Window<U, Integer, LinkedList<U>> window = Windows.lastNProcessOnInsert(1, tuple -> 0);
- Aggregate<U,Object,Integer> op = new Aggregate<U,Object,Integer>(window, processor);
+ public <J, U, K> TStream<J> joinLast(Function<T, K> keyer,
+ TStream<U> lastStream, Function<U, K> lastStreamKeyer, BiFunction<T, U, J> joiner) {
+ BiFunction<List<U>,K, Object> processor = Functions.synchronizedBiFunction((list, key) -> null);
+ Window<U, K, LinkedList<U>> window = Windows.lastNProcessOnInsert(1, lastStreamKeyer);
+ Aggregate<U,Object,K> op = new Aggregate<U,Object,K>(window, processor);
lastStream.pipe(op);
return this.map((tuple) -> {
- Partition<U, Integer, ? extends List<U>> part = window.getPartitions().get(0);
+ Partition<U, K, ? extends List<U>> part = window.getPartitions().get(keyer.apply(tuple));
if(part == null)
return null;
J ret;
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/f58e8f59/spi/topology/src/main/java/quarks/topology/spi/graph/TWindowTimeImpl.java
----------------------------------------------------------------------
diff --git a/spi/topology/src/main/java/quarks/topology/spi/graph/TWindowTimeImpl.java b/spi/topology/src/main/java/quarks/topology/spi/graph/TWindowTimeImpl.java
index 6194a05..a3ef0f3 100644
--- a/spi/topology/src/main/java/quarks/topology/spi/graph/TWindowTimeImpl.java
+++ b/spi/topology/src/main/java/quarks/topology/spi/graph/TWindowTimeImpl.java
@@ -104,23 +104,10 @@ public class TWindowTimeImpl<T, K> extends AbstractTWindow<T, K> {
}
/**
- * @param time the time to set
- */
- public void setTime(long time) {
- this.time = time;
- }
-
- /**
* @return the unit
*/
public TimeUnit getUnit() {
return unit;
}
- /**
- * @param unit the unit to set
- */
- public void setUnit(TimeUnit unit) {
- this.unit = unit;
- }
}
[2/3] incubator-quarks git commit: Added first round of tests for
join.
Posted by wc...@apache.org.
Added first round of tests for join.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quarks/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quarks/commit/678dc18b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quarks/tree/678dc18b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quarks/diff/678dc18b
Branch: refs/heads/master
Commit: 678dc18b17a6a7e150493b91b951fd2d6e0649de
Parents: 7aa29ef
Author: wcmarsha <wc...@us.ibm.com>
Authored: Thu Mar 31 21:05:35 2016 -0700
Committer: wcmarsha <wc...@us.ibm.com>
Committed: Tue Apr 5 10:08:52 2016 -0700
----------------------------------------------------------------------
.../java/quarks/test/topology/TStreamTest.java | 74 ++++++++++++++++++++
1 file changed, 74 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/678dc18b/api/topology/src/test/java/quarks/test/topology/TStreamTest.java
----------------------------------------------------------------------
diff --git a/api/topology/src/test/java/quarks/test/topology/TStreamTest.java b/api/topology/src/test/java/quarks/test/topology/TStreamTest.java
index 6b80d21..44210d3 100644
--- a/api/topology/src/test/java/quarks/test/topology/TStreamTest.java
+++ b/api/topology/src/test/java/quarks/test/topology/TStreamTest.java
@@ -45,6 +45,7 @@ import org.junit.Test;
import quarks.topology.TSink;
import quarks.topology.TStream;
+import quarks.topology.TWindow;
import quarks.topology.Topology;
import quarks.topology.tester.Condition;
@@ -678,6 +679,79 @@ public abstract class TStreamTest extends TopologyAbstractTest {
}
waitForCompletion(completer, executions);
}
+
+ @Test
+ public void testJoinWithWindow() throws Exception{
+ Topology t = newTopology();
+
+ List<Integer> ints = new ArrayList<>();
+ List<Integer> lookupInts = new ArrayList<>();
+
+ // Ints to populate the window
+ for(int i = 0; i < 100; i++){
+ ints.add(i);
+ }
+
+ // Ints to lookup partitions in window
+ for(int i = 0; i < 10; i++){
+ lookupInts.add(i);
+ }
+ TStream<Integer> intStream = t.collection(ints);
+
+ // Wait until the window is populated, and then submit tuples
+ TStream<Integer> lookupIntStream = t.source(() -> {
+ try {
+ Thread.sleep(500);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ return lookupInts;
+ });
+
+ TWindow<Integer, Integer> window = intStream.last(10, tuple -> tuple % 10);
+ TStream<Integer> joinsHappened = lookupIntStream.join(tuple -> tuple % 10, window, (number, partitionContents) -> {
+ assertTrue(partitionContents.size() == 10);
+ for(Integer element : partitionContents)
+ assertTrue(number % 10 == element % 10);
+
+ // Causes an error if two numbers map to the same partition, which shouldn't happen
+ partitionContents.clear();
+ return 0;
+ });
+
+ Condition<Long> tc = t.getTester().tupleCount(joinsHappened, 10);
+ complete(t, tc);
+ }
+
+ @Test
+ public void testJoinLastWithKeyer() throws Exception{
+ Topology t = newTopology();
+
+ List<Integer> ints = new ArrayList<>();
+ for(int i = 0; i < 100; i++){
+ ints.add(i);
+ }
+
+ TStream<Integer> intStream = t.collection(ints);
+
+ // Wait until the window is populated, and then submit tuples
+ TStream<Integer> lookupIntStream = t.source(() -> {
+ try {
+ Thread.sleep(500);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ return ints;
+ });
+
+ TStream<String> joinsHappened = lookupIntStream.joinLast(tuple -> tuple, intStream, tuple -> tuple, (a, b) -> {
+ assertTrue(a.equals(b));
+ return "0";
+ });
+
+ Condition<Long> tc = t.getTester().tupleCount(joinsHappened, 100);
+ complete(t, tc);
+ }
private void waitForCompletion(ExecutorCompletionService<Boolean> completer, int numtasks) throws ExecutionException {
int remainingTasks = numtasks;