You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@heron.apache.org by nw...@apache.org on 2018/11/26 16:09:18 UTC
[incubator-heron] branch master updated: [Java Streamlet API]
Support Abstractions on Streamlet Sources (#3118)
This is an automated email from the ASF dual-hosted git repository.
nwang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-heron.git
The following commit(s) were added to refs/heads/master by this push:
new 3619588 [Java Streamlet API] Support Abstractions on Streamlet Sources (#3118)
3619588 is described below
commit 3619588e858da436a5d8ff89083fe0e507305edf
Author: Eren Avsarogullari <er...@gmail.com>
AuthorDate: Mon Nov 26 16:09:13 2018 +0000
[Java Streamlet API] Support Abstractions on Streamlet Sources (#3118)
---
.../java/org/apache/heron/streamlet/Source.java | 2 --
.../streamlet/impl/sources/ComplexSource.java | 15 ++++--------
.../streamlet/impl/sources/StreamletSource.java | 14 +++++++++--
.../streamlet/impl/sources/SupplierSource.java | 15 +-----------
.../streamlet/impl/streamlets/SourceStreamlet.java | 2 --
.../heron/streamlet/impl/StreamletImplTest.java | 28 ++++++++++++++++++++++
6 files changed, 46 insertions(+), 30 deletions(-)
diff --git a/heron/api/src/java/org/apache/heron/streamlet/Source.java b/heron/api/src/java/org/apache/heron/streamlet/Source.java
index 320f3e4..b974471 100644
--- a/heron/api/src/java/org/apache/heron/streamlet/Source.java
+++ b/heron/api/src/java/org/apache/heron/streamlet/Source.java
@@ -16,8 +16,6 @@
* specific language governing permissions and limitations
* under the License.
*/
-
-
package org.apache.heron.streamlet;
import java.io.Serializable;
diff --git a/heron/api/src/java/org/apache/heron/streamlet/impl/sources/ComplexSource.java b/heron/api/src/java/org/apache/heron/streamlet/impl/sources/ComplexSource.java
index 3752b1c..237eb81 100644
--- a/heron/api/src/java/org/apache/heron/streamlet/impl/sources/ComplexSource.java
+++ b/heron/api/src/java/org/apache/heron/streamlet/impl/sources/ComplexSource.java
@@ -16,8 +16,6 @@
* specific language governing permissions and limitations
* under the License.
*/
-
-
package org.apache.heron.streamlet.impl.sources;
import java.io.Serializable;
@@ -38,10 +36,9 @@ import org.apache.heron.streamlet.impl.ContextImpl;
* to generate the next tuple.
*/
public class ComplexSource<R> extends StreamletSource {
+
private static final long serialVersionUID = -5086763670301450007L;
private Source<R> generator;
-
- private SpoutOutputCollector collector;
private State<Serializable, Serializable> state;
public ComplexSource(Source<R> generator) {
@@ -57,18 +54,16 @@ public class ComplexSource<R> extends StreamletSource {
@Override
public void open(Map<String, Object> map, TopologyContext topologyContext,
SpoutOutputCollector outputCollector) {
- collector = outputCollector;
+ super.open(map, topologyContext, outputCollector);
Context context = new ContextImpl(topologyContext, map, state);
generator.setup(context);
}
@Override
public void nextTuple() {
- Collection<R> val = generator.get();
- if (val != null) {
- for (R tuple : val) {
- collector.emit(new Values(tuple));
- }
+ Collection<R> tuples = generator.get();
+ if (tuples != null) {
+ tuples.forEach(tuple -> collector.emit(new Values(tuple)));
}
}
}
diff --git a/heron/api/src/java/org/apache/heron/streamlet/impl/sources/StreamletSource.java b/heron/api/src/java/org/apache/heron/streamlet/impl/sources/StreamletSource.java
index 16bbfb5..19c8ceb 100644
--- a/heron/api/src/java/org/apache/heron/streamlet/impl/sources/StreamletSource.java
+++ b/heron/api/src/java/org/apache/heron/streamlet/impl/sources/StreamletSource.java
@@ -16,16 +16,17 @@
* specific language governing permissions and limitations
* under the License.
*/
-
-
package org.apache.heron.streamlet.impl.sources;
import java.io.Serializable;
+import java.util.Map;
import org.apache.heron.api.spout.BaseRichSpout;
+import org.apache.heron.api.spout.SpoutOutputCollector;
import org.apache.heron.api.state.State;
import org.apache.heron.api.topology.IStatefulComponent;
import org.apache.heron.api.topology.OutputFieldsDeclarer;
+import org.apache.heron.api.topology.TopologyContext;
import org.apache.heron.api.tuple.Fields;
/**
@@ -38,12 +39,21 @@ public abstract class StreamletSource extends BaseRichSpout
private static final long serialVersionUID = 8583965332619565343L;
private static final String OUTPUT_FIELD_NAME = "output";
+ protected SpoutOutputCollector collector;
+
@Override
public void initState(State<Serializable, Serializable> state) { }
@Override
public void preSave(String checkpointId) { }
+ @SuppressWarnings("rawtypes")
+ @Override
+ public void open(Map<String, Object> map, TopologyContext topologyContext,
+ SpoutOutputCollector outputCollector) {
+ collector = outputCollector;
+ }
+
/**
* The sources implementing streamlet functionality have some properties.
* 1. They all output only one stream
diff --git a/heron/api/src/java/org/apache/heron/streamlet/impl/sources/SupplierSource.java b/heron/api/src/java/org/apache/heron/streamlet/impl/sources/SupplierSource.java
index 55efaa3..d60a34f 100644
--- a/heron/api/src/java/org/apache/heron/streamlet/impl/sources/SupplierSource.java
+++ b/heron/api/src/java/org/apache/heron/streamlet/impl/sources/SupplierSource.java
@@ -16,14 +16,8 @@
* specific language governing permissions and limitations
* under the License.
*/
-
-
package org.apache.heron.streamlet.impl.sources;
-import java.util.Map;
-
-import org.apache.heron.api.spout.SpoutOutputCollector;
-import org.apache.heron.api.topology.TopologyContext;
import org.apache.heron.api.tuple.Values;
import org.apache.heron.streamlet.SerializableSupplier;
@@ -33,21 +27,14 @@ import org.apache.heron.streamlet.SerializableSupplier;
* to generate the next tuple.
*/
public class SupplierSource<R> extends StreamletSource {
+
private static final long serialVersionUID = 6476611751545430216L;
private SerializableSupplier<R> supplier;
- private SpoutOutputCollector collector;
-
public SupplierSource(SerializableSupplier<R> supplier) {
this.supplier = supplier;
}
- @SuppressWarnings("rawtypes")
- @Override
- public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector outputCollector) {
- collector = outputCollector;
- }
-
@Override
public void nextTuple() {
collector.emit(new Values(supplier.get()));
diff --git a/heron/api/src/java/org/apache/heron/streamlet/impl/streamlets/SourceStreamlet.java b/heron/api/src/java/org/apache/heron/streamlet/impl/streamlets/SourceStreamlet.java
index dbc6f18..edba615 100644
--- a/heron/api/src/java/org/apache/heron/streamlet/impl/streamlets/SourceStreamlet.java
+++ b/heron/api/src/java/org/apache/heron/streamlet/impl/streamlets/SourceStreamlet.java
@@ -16,8 +16,6 @@
* specific language governing permissions and limitations
* under the License.
*/
-
-
package org.apache.heron.streamlet.impl.streamlets;
import java.util.Set;
diff --git a/heron/api/tests/java/org/apache/heron/streamlet/impl/StreamletImplTest.java b/heron/api/tests/java/org/apache/heron/streamlet/impl/StreamletImplTest.java
index 7a76dfc..33f3a9c 100644
--- a/heron/api/tests/java/org/apache/heron/streamlet/impl/StreamletImplTest.java
+++ b/heron/api/tests/java/org/apache/heron/streamlet/impl/StreamletImplTest.java
@@ -19,6 +19,7 @@
package org.apache.heron.streamlet.impl;
import java.util.Arrays;
+import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -42,6 +43,7 @@ import org.apache.heron.streamlet.IStreamletRichOperator;
import org.apache.heron.streamlet.IStreamletWindowOperator;
import org.apache.heron.streamlet.SerializableConsumer;
import org.apache.heron.streamlet.SerializableTransformer;
+import org.apache.heron.streamlet.Source;
import org.apache.heron.streamlet.Streamlet;
import org.apache.heron.streamlet.WindowConfig;
import org.apache.heron.streamlet.impl.streamlets.ConsumerStreamlet;
@@ -51,6 +53,7 @@ import org.apache.heron.streamlet.impl.streamlets.FlatMapStreamlet;
import org.apache.heron.streamlet.impl.streamlets.JoinStreamlet;
import org.apache.heron.streamlet.impl.streamlets.MapStreamlet;
import org.apache.heron.streamlet.impl.streamlets.ReduceByKeyAndWindowStreamlet;
+import org.apache.heron.streamlet.impl.streamlets.SourceStreamlet;
import org.apache.heron.streamlet.impl.streamlets.SpoutStreamlet;
import org.apache.heron.streamlet.impl.streamlets.SupplierStreamlet;
import org.apache.heron.streamlet.impl.streamlets.TransformStreamlet;
@@ -91,6 +94,12 @@ public class StreamletImplTest {
}
@Test
+ public void testSourceStreamlet() {
+ Streamlet<String> streamlet = builder.newSource(new TestSource());
+ assertTrue(streamlet instanceof SourceStreamlet);
+ }
+
+ @Test
public void testSpoutStreamlet() {
TestSpout spout = new TestSpout();
Streamlet<Double> streamlet = builder.newSource(spout);
@@ -498,4 +507,23 @@ public class StreamletImplTest {
}
}
+ private class TestSource implements Source<String> {
+
+ private List<String> list;
+ @Override
+ public void setup(Context context) {
+ list = Arrays.asList("aa", "bb", "cc");
+ }
+
+ @Override
+ public Collection<String> get() {
+ return list;
+ }
+
+ @Override
+ public void cleanup() {
+ list.clear();
+ }
+ }
+
}