diff --git a/examples/src/scala/BUILD b/examples/src/scala/BUILD
index d03d9a8..786ac2a 100644
--- a/examples/src/scala/BUILD
+++ b/examples/src/scala/BUILD
@@ -8,6 +8,7 @@
     deps = [
         "@org_apache_commons_commons_lang3//jar",
         "//heron/api/src/java:api-java",
+        "//heron/api/src/java:api-java-low-level",
         "//heron/api/src/scala:api-scala",
         "//third_party/java:kryo"
     ],
diff --git a/heron/api/src/java/org/apache/heron/streamlet/Builder.java b/heron/api/src/java/org/apache/heron/streamlet/Builder.java
index 8e7c6e7..d6e4f70 100644
--- a/heron/api/src/java/org/apache/heron/streamlet/Builder.java
+++ b/heron/api/src/java/org/apache/heron/streamlet/Builder.java
@@ -20,6 +20,7 @@
 
 package org.apache.heron.streamlet;
 
+import org.apache.heron.api.spout.IRichSpout;
 import org.apache.heron.streamlet.impl.BuilderImpl;
 
 /**
@@ -45,4 +46,12 @@
    * @return
    */
   <R> Streamlet<R> newSource(Source<R> generator);
+
+  /**
+   * Creates a new Streamlet using the provided spout
+   * @param spout The spout that emits the tuples of the streamlet
+   * @param <R>
+   * @return
+   */
+  <R> Streamlet<R> newSource(IRichSpout spout);
 }
diff --git a/heron/api/src/java/org/apache/heron/streamlet/impl/BuilderImpl.java b/heron/api/src/java/org/apache/heron/streamlet/impl/BuilderImpl.java
index 0eb06d9..61715e8 100644
--- a/heron/api/src/java/org/apache/heron/streamlet/impl/BuilderImpl.java
+++ b/heron/api/src/java/org/apache/heron/streamlet/impl/BuilderImpl.java
@@ -24,6 +24,7 @@
 import java.util.List;
 import java.util.Set;
 
+import org.apache.heron.api.spout.IRichSpout;
 import org.apache.heron.api.topology.TopologyBuilder;
 import org.apache.heron.streamlet.Builder;
 import org.apache.heron.streamlet.SerializableSupplier;
@@ -47,7 +48,6 @@
   public <R> Streamlet<R> newSource(SerializableSupplier<R> supplier) {
     StreamletUtils.require(supplier != null, "supplier must not be null.");
     StreamletImpl<R> retval = StreamletImpl.createSupplierStreamlet(supplier);
-    retval.setNumPartitions(1);
     sources.add(retval);
     return retval;
   }
@@ -56,7 +56,14 @@
   public <R> Streamlet<R> newSource(Source<R> generator) {
     StreamletUtils.require(generator != null, "source must not be null.");
     StreamletImpl<R> retval = StreamletImpl.createGeneratorStreamlet(generator);
-    retval.setNumPartitions(1);
+    sources.add(retval);
+    return retval;
+  }
+
+  @Override
+  public <R> Streamlet<R> newSource(IRichSpout spout) {
+    StreamletUtils.require(spout != null, "spout must not be null.");
+    StreamletImpl<R> retval = StreamletImpl.createSpoutStreamlet(spout);
     sources.add(retval);
     return retval;
   }
diff --git a/heron/api/src/java/org/apache/heron/streamlet/impl/StreamletImpl.java b/heron/api/src/java/org/apache/heron/streamlet/impl/StreamletImpl.java
index 066dc74..a05040b 100644
--- a/heron/api/src/java/org/apache/heron/streamlet/impl/StreamletImpl.java
+++ b/heron/api/src/java/org/apache/heron/streamlet/impl/StreamletImpl.java
@@ -26,6 +26,7 @@
 import java.util.Set;
 import java.util.logging.Logger;
 
+import org.apache.heron.api.spout.IRichSpout;
 import org.apache.heron.api.topology.TopologyBuilder;
 import org.apache.heron.streamlet.IStreamletOperator;
 import org.apache.heron.streamlet.JoinType;
@@ -54,6 +55,7 @@
 import org.apache.heron.streamlet.impl.streamlets.RemapStreamlet;
 import org.apache.heron.streamlet.impl.streamlets.SinkStreamlet;
 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;
 import org.apache.heron.streamlet.impl.streamlets.UnionStreamlet;
@@ -115,6 +117,7 @@
     REMAP("remap"),
     SINK("sink"),
     SOURCE("generator"),
+    SPOUT("spout"),
     SUPPLIER("supplier"),
     TRANSFORM("transform"),
     UNION("union");
@@ -264,6 +267,14 @@
   }
 
   /**
+   * Create a Streamlet based on a Spout object
+   * @param spout The Spout function to generate the elements
+   */
+  static <T> StreamletImpl<T> createSpoutStreamlet(IRichSpout spout) {
+    return new SpoutStreamlet<T>(spout);
+  }
+
+  /**
    * Return a new Streamlet by applying mapFn to each element of this Streamlet
    * @param mapFn The Map Function that should be applied to each element
   */
diff --git a/heron/api/src/java/org/apache/heron/streamlet/impl/streamlets/SpoutStreamlet.java b/heron/api/src/java/org/apache/heron/streamlet/impl/streamlets/SpoutStreamlet.java
new file mode 100644
index 0000000..5bb12cf
--- /dev/null
+++ b/heron/api/src/java/org/apache/heron/streamlet/impl/streamlets/SpoutStreamlet.java
@@ -0,0 +1,47 @@
+/**
+ * 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.heron.streamlet.impl.streamlets;
+
+import java.util.Set;
+
+import org.apache.heron.api.spout.IRichSpout;
+import org.apache.heron.api.topology.TopologyBuilder;
+import org.apache.heron.streamlet.impl.StreamletImpl;
+
+/**
+ * SpoutStreamlet is a quick way of creating a Streamlet
+ * from an user supplied Spout object. The spout is the
+ * source of all tuples for this Streamlet.
+ */
+public class SpoutStreamlet<R> extends StreamletImpl<R> {
+  private IRichSpout spout;
+
+  public SpoutStreamlet(IRichSpout spout) {
+    this.spout = spout;
+    setNumPartitions(1);
+  }
+
+  @Override
+  public boolean doBuild(TopologyBuilder bldr, Set<String> stageNames) {
+    setDefaultNameIfNone(StreamletNamePrefix.SPOUT, stageNames);
+    bldr.setSpout(getName(), spout, getNumPartitions());
+    return true;
+  }
+}
diff --git a/heron/api/src/scala/org/apache/heron/streamlet/scala/Builder.scala b/heron/api/src/scala/org/apache/heron/streamlet/scala/Builder.scala
index 0e5f13e..b299f52 100644
--- a/heron/api/src/scala/org/apache/heron/streamlet/scala/Builder.scala
+++ b/heron/api/src/scala/org/apache/heron/streamlet/scala/Builder.scala
@@ -18,6 +18,7 @@
  */
 package org.apache.heron.streamlet.scala
 
+import org.apache.heron.api.spout.IRichSpout
 import org.apache.heron.streamlet.scala.impl.BuilderImpl
 
 /**
@@ -49,7 +50,14 @@
     * Creates a new Streamlet using the underlying generator
     *
     * @param generator The generator that generates the tuples of the streamlet
-    * @return  a Streamlet representation of the source object
+    * @return a Streamlet representation of the source object
     */
   def newSource[R](generator: Source[R]): Streamlet[R]
+
+  /**
+   * Creates a new Streamlet using the provided spout
+   * @param spout The spout that emits the tuples of the streamlet
+   * @return a Streamlet representation of the spout object
+   */
+  def newSource[R](spout: IRichSpout): Streamlet[R]
 }
diff --git a/heron/api/src/scala/org/apache/heron/streamlet/scala/impl/BuilderImpl.scala b/heron/api/src/scala/org/apache/heron/streamlet/scala/impl/BuilderImpl.scala
index 0f012e4..6020320 100644
--- a/heron/api/src/scala/org/apache/heron/streamlet/scala/impl/BuilderImpl.scala
+++ b/heron/api/src/scala/org/apache/heron/streamlet/scala/impl/BuilderImpl.scala
@@ -18,6 +18,7 @@
  */
 package org.apache.heron.streamlet.scala.impl
 
+import org.apache.heron.api.spout.IRichSpout
 import org.apache.heron.api.topology.TopologyBuilder
 import org.apache.heron.streamlet.impl.{BuilderImpl => JavaBuilderImpl}
 
@@ -39,6 +40,11 @@
     StreamletImpl.fromJavaStreamlet[R](newJavaStreamlet)
   }
 
+  override def newSource[R](spout: IRichSpout): Streamlet[R] = {
+    val newJavaStreamlet = builder.newSource[R](spout)
+    StreamletImpl.fromJavaStreamlet[R](newJavaStreamlet)
+  }
+
   def build(): TopologyBuilder =
     builder.asInstanceOf[JavaBuilderImpl].build()
 
diff --git a/heron/api/tests/java/org/apache/heron/resource/TestSpout.java b/heron/api/tests/java/org/apache/heron/resource/TestSpout.java
new file mode 100644
index 0000000..49025bd
--- /dev/null
+++ b/heron/api/tests/java/org/apache/heron/resource/TestSpout.java
@@ -0,0 +1,81 @@
+/**
+ * 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.heron.resource;
+
+
+import java.util.List;
+import java.util.Map;
+
+import org.junit.Ignore;
+
+import org.apache.heron.api.spout.BaseRichSpout;
+import org.apache.heron.api.spout.SpoutOutputCollector;
+import org.apache.heron.api.topology.OutputFieldsDeclarer;
+import org.apache.heron.api.topology.TopologyContext;
+import org.apache.heron.api.tuple.Fields;
+import org.apache.heron.api.tuple.Values;
+
+/**
+ * A Spout used for unit test, it will:
+ * 1. It will emit EMIT_COUNT of tuples with MESSAGE_ID.
+ * 2. The tuples are declared by outputFieldsDeclarer in fields "word"
+ */
+
+@Ignore
+public class TestSpout extends BaseRichSpout {
+  private static final int EMIT_COUNT = 10;
+  private static final String MESSAGE_ID = "MESSAGE_ID";
+
+  private final String[] toSend = new String[]{"A", "B"};
+  private SpoutOutputCollector outputCollector;
+  private int emitted = 0;
+
+  @Override
+  public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
+    outputFieldsDeclarer.declare(new Fields("word"));
+  }
+
+  @Override
+  public void open(
+      Map<String, Object> map,
+      TopologyContext topologyContext,
+      SpoutOutputCollector spoutOutputCollector) {
+    this.outputCollector = spoutOutputCollector;
+  }
+
+  @Override
+  public void close() {
+  }
+
+  @Override
+  public void nextTuple() {
+    // It will emit A, B, A, B, A, B, A, B, A, B
+    if (emitted < EMIT_COUNT) {
+      String word = toSend[emitted % toSend.length];
+      emit(outputCollector, new Values(word), MESSAGE_ID, emitted++);
+    }
+  }
+
+  protected void emit(SpoutOutputCollector collector,
+                      List<Object> tuple, Object messageId, int emittedCount) {
+    collector.emit(tuple, messageId);
+  }
+}
+
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 8a0b22d..a9b15bd 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
@@ -32,6 +32,7 @@
 import org.apache.heron.common.basics.ByteAmount;
 import org.apache.heron.resource.TestBasicBolt;
 import org.apache.heron.resource.TestBolt;
+import org.apache.heron.resource.TestSpout;
 import org.apache.heron.resource.TestWindowBolt;
 import org.apache.heron.streamlet.Config;
 import org.apache.heron.streamlet.Context;
@@ -49,6 +50,7 @@
 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.SpoutStreamlet;
 import org.apache.heron.streamlet.impl.streamlets.SupplierStreamlet;
 import org.apache.heron.streamlet.impl.streamlets.TransformStreamlet;
 import org.apache.heron.streamlet.impl.streamlets.UnionStreamlet;
@@ -86,6 +88,13 @@
   }
 
   @Test
+  public void testSpoutStreamlet() throws Exception {
+    TestSpout spout = new TestSpout();
+    Streamlet<Double> streamlet = StreamletImpl.createSpoutStreamlet(spout);
+    assertTrue(streamlet instanceof SpoutStreamlet);
+  }
+
+  @Test
   @SuppressWarnings("unchecked")
   public void testMapStreamlet() throws Exception {
     Streamlet<Double> baseStreamlet = StreamletImpl.createSupplierStreamlet(() -> Math.random());
diff --git a/heron/api/tests/scala/org/apache/heron/resource/TestSpout.scala b/heron/api/tests/scala/org/apache/heron/resource/TestSpout.scala
new file mode 100644
index 0000000..e0b7e08
--- /dev/null
+++ b/heron/api/tests/scala/org/apache/heron/resource/TestSpout.scala
@@ -0,0 +1,79 @@
+/**
+ * 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.heron.resource
+
+
+import java.util.{List => JList}
+import java.util.{Map => JMap}
+
+import org.junit.Ignore
+
+import org.apache.heron.api.spout.BaseRichSpout
+import org.apache.heron.api.spout.SpoutOutputCollector
+import org.apache.heron.api.topology.OutputFieldsDeclarer
+import org.apache.heron.api.topology.TopologyContext
+import org.apache.heron.api.tuple.Fields
+import org.apache.heron.api.tuple.Values
+
+/**
+ * A Spout used for unit test, it will:
+ * 1. It will emit EMIT_COUNT of tuples with MESSAGE_ID.
+ * 2. The tuples are declared by outputFieldsDeclarer in fields "word"
+ */
+
+@Ignore
+class TestSpout extends BaseRichSpout {
+  private val EMIT_COUNT = 10
+  private val MESSAGE_ID = "MESSAGE_ID"
+
+  private val toSend = Array("A", "B")
+  private var outputCollector: SpoutOutputCollector = _
+  private var emitted = 0
+
+  override def declareOutputFields(outputFieldsDeclarer: OutputFieldsDeclarer ): Unit = {
+    outputFieldsDeclarer.declare(new Fields("word"))
+  }
+
+  override def open(
+      conf: JMap[String, Object],
+      topologyContext: TopologyContext,
+      spoutOutputCollector: SpoutOutputCollector): Unit = {
+    this.outputCollector = spoutOutputCollector
+  }
+
+  override def close(): Unit = {
+  }
+
+  override def nextTuple(): Unit = {
+    // It will emit A, B, A, B, A, B, A, B, A, B
+    if (emitted < EMIT_COUNT) {
+      val word = toSend(emitted % toSend.length)
+      emit(outputCollector, new Values(word), MESSAGE_ID, emitted)
+      emitted = emitted + 1
+    }
+  }
+
+  def emit(collector: SpoutOutputCollector,
+           tuple: JList[Object],
+           messageId: Object,
+           emittedCount: Int): Unit = {
+    collector.emit(tuple, messageId)
+  }
+}
diff --git a/heron/api/tests/scala/org/apache/heron/streamlet/scala/impl/BuilderImplTest.scala b/heron/api/tests/scala/org/apache/heron/streamlet/scala/impl/BuilderImplTest.scala
index cc04caa..3d8048b 100644
--- a/heron/api/tests/scala/org/apache/heron/streamlet/scala/impl/BuilderImplTest.scala
+++ b/heron/api/tests/scala/org/apache/heron/streamlet/scala/impl/BuilderImplTest.scala
@@ -18,12 +18,14 @@
  */
 package org.apache.heron.streamlet.scala.impl
 
+import java.util.{Map => JMap}
+import scala.collection.JavaConverters._
 import scala.collection.mutable.ListBuffer
 
 import org.junit.Assert.assertEquals
 
+import org.apache.heron.resource.TestSpout
 import org.apache.heron.streamlet.Context
-
 import org.apache.heron.streamlet.scala.{Builder, Streamlet, Source}
 import org.apache.heron.streamlet.scala.common.BaseFunSuite
 
@@ -69,4 +71,16 @@
     override def cleanup(): Unit = numbers.clear()
   }
 
+  test(
+    "BuilderImpl should support streamlet generation from a user defined spout") {
+    val spout = new TestSpout
+    val spoutStreamletObj = Builder.newBuilder
+      .newSource(spout)
+      .setName("Spout_Streamlet_1")
+      .setNumPartitions(20)
+
+    assert(spoutStreamletObj.isInstanceOf[Streamlet[_]])
+    assertEquals("Spout_Streamlet_1", spoutStreamletObj.getName)
+    assertEquals(20, spoutStreamletObj.getNumPartitions)
+  }
 }