diff --git a/.gitignore b/.gitignore
index 3cf0208..294c718 100644
--- a/.gitignore
+++ b/.gitignore
@@ -11,4 +11,5 @@
 
 #intellij
 .idea/
-.iml
+*.iml
+*.iws
diff --git a/bin/samoa b/bin/samoa
index b34f65b..0ace74b 100755
--- a/bin/samoa
+++ b/bin/samoa
@@ -4,7 +4,7 @@
 # #%L
 # SAMOA
 # %%
-# Copyright (C) 2013 Yahoo! Inc.
+# Copyright (C) 2014 - 2015 Apache Software Foundation
 # %%
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -281,6 +281,66 @@
                --kryo_register=$BASE_DIR/$KRYO_REGISTER_FILE --pi_per_container=$PI_PER_CONTAINER \
 	       --samoa_hdfs_dir=$HDFS_SAMOA_HOME
 
+elif [ $PLATFORM = 'FLINK' ]; then
+
+        echo "Deploying to $PLATFORM"
+        if [ -z $FLINK_HOME ];then
+            echo "FLINK_HOME is not set!"
+            echo "Please set FLINK_HOME to point to your Flink installation"
+            exit -1
+        fi
+
+        if [ ! -f $2 ];then
+            echo "$2 does not exist!"
+            echo "Please use a valid jar file for Flink execution"
+            exit -1
+        fi
+
+        FLINK_EXEC="$FLINK_HOME/bin/flink"
+
+        SAMOA_FLINK_PROPERTIES="samoa-flink.properties"
+        MODE_OPTION="samoa.flink.mode"
+#        NUM_WORKER_OPTION="samoa.flink.numWorker"
+
+        VALUE=""
+        getvalue()
+        {
+            VALUE=`sed '/^\#/d' $BASE_DIR/$SAMOA_FLINK_PROPERTIES | grep "$1" | tail -n 1 | cut -d "=" -f2- | sed 's/^[[:space:]]*//;s/[[:space:]]*$//'`
+        }
+        
+#        getvalue "$NUM_WORKER_OPTION"
+#        NUM_WORKER="$VALUE"
+
+        getvalue "$MODE_OPTION"
+        MODE_ARG="$VALUE"
+
+        COMPLETE_ARG=""
+        COUNTER=0
+        for var in "$@"
+        do
+            COUNTER=`expr $COUNTER + 1`
+            if [ $COUNTER -gt 2 ];then
+                COMPLETE_ARG="$COMPLETE_ARG $var"
+            fi
+        done
+
+        DEPLOYABLE=$JAR_PATH
+        echo "$DEPLOYABLE"
+        if [ "$MODE_ARG" = "cluster" ]; then
+				FLINK_MASTER_OPTION="samoa.flink.flinkMaster"
+				PORT_OPTION="samoa.flink.port"
+
+				getvalue "$FLINK_MASTER_OPTION"
+				FLINK_MASTER_OPTION="$VALUE"
+
+				getvalue "$PORT_OPTION"
+				PORT_OPTION="$VALUE"
+                $FLINK_EXEC run -m $FLINK_MASTER_OPTION:$PORT_OPTION $DEPLOYABLE  $COMPLETE_ARG
+
+        elif [ "$MODE_ARG" = "local" ]; then
+                $FLINK_EXEC run $DEPLOYABLE $COMPLETE_ARG
+        fi                                                    
+
 elif [ $PLATFORM = 'THREADS' ]; then
 	
 	echo "Deploying to LOCAL with MULTITHREADING."
diff --git a/bin/samoa-flink.properties b/bin/samoa-flink.properties
new file mode 100644
index 0000000..b9f56c0
--- /dev/null
+++ b/bin/samoa-flink.properties
@@ -0,0 +1,35 @@
+###
+# #%L
+# SAMOA
+# %%
+# Copyright (C) 2014 - 2015 Apache Software Foundation
+# %%
+# Licensed 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.
+# #L%
+###
+
+# SAMOA Flink properties file
+# This file contains specific configurations for SAMOA deployment in Flink platform
+
+# samoa.flink.mode corresponds to the execution mode of a Task in Flink
+# possible values:
+# 1. local   - to run the task in local StreamingEnvironment
+# 2. cluster - to run the task in the specified cluster
+samoa.flink.mode=cluster
+
+#in case samoa.flink.mode equals "cluster", then the user has to set up also the following parameters:
+# @samoa.flink.flinkMaster: the IP address of the cluster
+# @samoa.flink.port : the port
+samoa.flink.flinkMaster=127.0.0.1
+samoa.flink.port=6123
+
diff --git a/pom.xml b/pom.xml
index 622d952..819a13c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -88,6 +88,15 @@
             </modules>
         </profile>
         <profile>
+            <id>flink</id>
+            <modules>
+                <module>samoa-instances</module>
+                <module>samoa-api</module>
+                <module>samoa-flink</module>
+                <module>samoa-test</module>
+            </modules>
+        </profile>
+        <profile>
             <id>samza</id>
             <modules>
                 <module>samoa-instances</module>
@@ -104,6 +113,7 @@
                 <module>samoa-local</module>
                 <module>samoa-threads</module>
                 <module>samoa-storm</module>
+                <module>samoa-flink</module>
                 <module>samoa-s4</module>
                 <module>samoa-samza</module>
                 <module>samoa-test</module>
@@ -130,6 +140,7 @@
         <miniball.version>1.0.3</miniball.version>
         <s4.version>0.6.0-incubating</s4.version>
         <samza.version>0.7.0</samza.version>
+        <flink.version>0.9.0-milestone-1</flink.version>
         <slf4j-log4j12.version>1.7.2</slf4j-log4j12.version>
         <slf4j-simple.version>1.7.5</slf4j-simple.version>
         <maven-surefire-plugin.version>2.18</maven-surefire-plugin.version>
@@ -211,6 +222,7 @@
                         <root>samoa-local</root>
                         <root>samoa-storm</root>
                         <root>samoa-s4</root>
+                        <root>samoa-flink</root>
                         <root>samoa-samza</root>
                         <root>bin</root>
                     </roots>
diff --git a/samoa-flink/pom.xml b/samoa-flink/pom.xml
new file mode 100644
index 0000000..f00fe3c
--- /dev/null
+++ b/samoa-flink/pom.xml
@@ -0,0 +1,134 @@
+<!--
+  #%L
+  SAMOA
+  %%
+  Copyright (C) 2013 Yahoo! Inc.
+  %%
+  Licensed 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.
+  #L%
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+            <kryo.version>2.24.0</kryo.version>
+    </properties>
+      <repositories>
+                <repository>
+                        <id>apache.snapshots</id>
+                        <name>Apache Development Snapshot Repository</name>
+                        <url>https://repository.apache.org/content/repositories/snapshots/</url>
+                        <releases>
+                                <enabled>false</enabled>
+                        </releases>
+                        <snapshots>
+                                <enabled>true</enabled>
+                        </snapshots>
+                </repository>
+        </repositories>
+    <name>samoa-flink</name>
+    <description>Flink engine for SAMOA</description>
+
+    <artifactId>samoa-flink</artifactId>
+    <parent>
+        <groupId>com.yahoo.labs.samoa</groupId>
+        <artifactId>samoa</artifactId>
+        <version>0.3.0-SNAPSHOT</version>
+    </parent>
+    
+    
+
+    <dependencies>
+        <dependency>
+            <groupId>com.yahoo.labs.samoa</groupId>
+            <artifactId>samoa-api</artifactId>
+            <version>${project.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.esotericsoftware.kryo</groupId>
+                    <artifactId>kryo</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-log4j12</artifactId>
+            <version>${slf4j-log4j12.version}</version>
+        </dependency>
+  		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-streaming-core</artifactId>
+			<version>${flink.version}</version>
+            <!--<scope>provided</scope>-->
+		</dependency>
+        <dependency>
+            <groupId>com.esotericsoftware.kryo</groupId>
+            <artifactId>kryo</artifactId>
+            <version>${kryo.version}</version>
+        </dependency>
+    </dependencies>
+
+
+    <build>
+        <plugins>
+            <!-- Flink assembly -->
+            <plugin>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <version>${maven-assembly-plugin.version}</version>
+                <configuration>
+                    <finalName>SAMOA-Flink-${project.version}</finalName>
+                    <appendAssemblyId>false</appendAssemblyId>
+                    <attach>false</attach>
+                    <outputDirectory>../target</outputDirectory>
+                    <descriptorRefs>
+                        <descriptorRef>jar-with-dependencies</descriptorRef>
+                    </descriptorRefs>
+                    <archive>
+                        <manifestEntries>
+                            <!--<Bundle-Version>${parsedVersion.osgiVersion}</Bundle-Version>-->
+                            <Bundle-Description>${project.description}</Bundle-Description>
+                            <Implementation-Version>${project.version}</Implementation-Version>
+                            <Implementation-Vendor>Yahoo Labs</Implementation-Vendor>
+                            <Implementation-Vendor-Id>SAMOA</Implementation-Vendor-Id>
+                        </manifestEntries>
+                        <manifest>
+                            <addClasspath>true</addClasspath>
+                            <mainClass>com.yahoo.labs.flink.FlinkDoTask</mainClass>
+                        </manifest>
+                    </archive>
+                </configuration>
+                <executions>
+                    <execution>
+                        <id>make-assembly</id> <!-- this is used for inheritance merges -->
+                        <phase>package</phase>
+                        <goals>
+                            <goal>single</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>${maven-surefire-plugin.version}</version>
+                <configuration>
+                    <argLine>-Xmx1G</argLine>
+                    <redirectTestOutputToFile>false</redirectTestOutputToFile>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+    
+</project>
diff --git a/samoa-flink/src/main/java/com/yahoo/labs/flink/FlinkDoTask.java b/samoa-flink/src/main/java/com/yahoo/labs/flink/FlinkDoTask.java
new file mode 100644
index 0000000..6069de9
--- /dev/null
+++ b/samoa-flink/src/main/java/com/yahoo/labs/flink/FlinkDoTask.java
@@ -0,0 +1,87 @@
+package com.yahoo.labs.flink;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed 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.
+ * #L%
+ */
+
+import com.github.javacliparser.ClassOption;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.yahoo.labs.flink.com.yahoo.labs.flink.helpers.CircleDetection;
+import com.yahoo.labs.flink.com.yahoo.labs.flink.helpers.Utils;
+import com.yahoo.labs.flink.topology.impl.FlinkComponentFactory;
+import com.yahoo.labs.flink.topology.impl.FlinkProcessingItem;
+import com.yahoo.labs.flink.topology.impl.FlinkStream;
+import com.yahoo.labs.flink.topology.impl.FlinkTopology;
+import com.yahoo.labs.samoa.tasks.Task;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+
+
+/**
+ * Main class to run a SAMOA on Apache Flink
+ */
+public class FlinkDoTask {
+
+	private static final Logger logger = LoggerFactory.getLogger(FlinkDoTask.class);
+
+
+	public static void main(String[] args) throws Exception {
+		List<String> tmpArgs = new ArrayList<String>(Arrays.asList(args));
+
+		args = tmpArgs.toArray(new String[0]);
+
+		// Init Task
+		StringBuilder cliString = new StringBuilder();
+		for (int i = 0; i < args.length; i++) {
+			cliString.append(" ").append(args[i]);
+		}
+		logger.debug("Command line string = {}", cliString.toString());
+		System.out.println("Command line string = " + cliString.toString());
+
+		Task task;
+		try {
+			task = ClassOption.cliStringToObject(cliString.toString(), Task.class, null);
+			logger.debug("Successfully instantiating {}", task.getClass().getCanonicalName());
+		} catch (Exception e) {
+			logger.error("Failed to initialize the task: ", e);
+			System.out.println("Failed to initialize the task: " + e);
+			return;
+		}
+		
+		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+		task.setFactory(new FlinkComponentFactory(env));
+		task.init();
+		
+		logger.debug("Building Flink topology...");
+		((FlinkTopology) task.getTopology()).build();
+		
+		logger.debug("Submitting the job...");
+		env.execute();
+
+	}
+
+
+	
+
+}
diff --git a/samoa-flink/src/main/java/com/yahoo/labs/flink/com/yahoo/labs/flink/helpers/CircleDetection.java b/samoa-flink/src/main/java/com/yahoo/labs/flink/com/yahoo/labs/flink/helpers/CircleDetection.java
new file mode 100644
index 0000000..a832ee9
--- /dev/null
+++ b/samoa-flink/src/main/java/com/yahoo/labs/flink/com/yahoo/labs/flink/helpers/CircleDetection.java
@@ -0,0 +1,99 @@
+package com.yahoo.labs.flink.com.yahoo.labs.flink.helpers;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed 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.
+ * #L%
+ */
+
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Stack;
+
+/**
+ * This class contains all logic needed in order to mark circles in job graphs explicitly such as 
+ * in the case of Apache Flink. A circle is defined as a list of node ids ordered in topological 
+ * (DFS) order.
+ * 
+ */
+public class CircleDetection {
+	private int[] index;
+	private int[] lowLink;
+	private int counter;
+	private Stack<Integer> stack;
+	private List<List<Integer>> scc;
+	List<Integer>[] graph;
+
+
+	public CircleDetection() {
+		stack = new Stack<Integer>();
+		scc = new ArrayList<>();
+	}
+
+	public List<List<Integer>> getCircles(List<Integer>[] adjacencyList) {
+		graph = adjacencyList;
+		index = new int[adjacencyList.length];
+		lowLink = new int[adjacencyList.length];
+		counter = 0;
+
+		//initialize index and lowLink as "undefined"(=-1)
+		for (int j = 0; j < graph.length; j++) {
+			index[j] = -1;
+			lowLink[j] = -1;
+		}
+		for (int v = 0; v < graph.length; v++) {
+			if (index[v] == -1) { //undefined.
+				findSCC(v);
+			}
+		}
+		return scc;
+	}
+
+	private void findSCC(int node) {
+		index[node] = counter;
+		lowLink[node] = counter;
+		counter++;
+		stack.push(node);
+
+		for (int neighbor : graph[node]) {
+			if (index[neighbor] == -1) {
+				findSCC(neighbor);
+				lowLink[node] = Math.min(lowLink[node], lowLink[neighbor]);
+			} else if (stack.contains(neighbor)) { //if neighbor has been already visited
+				lowLink[node] = Math.min(lowLink[node], index[neighbor]);
+				List<Integer> sccComponent = new ArrayList<Integer>();
+				int w;
+				do {
+					w = stack.pop();
+					sccComponent.add(w);
+				} while (neighbor != w);
+				//add neighbor again, just in case it is a member of another circle 
+				stack.add(neighbor); 
+				scc.add(sccComponent);
+			}
+
+		}
+		if (lowLink[node] == index[node]) {
+			int w;
+			do {
+				w = stack.pop();
+			} while (node != w);
+		}
+	}
+
+}
\ No newline at end of file
diff --git a/samoa-flink/src/main/java/com/yahoo/labs/flink/com/yahoo/labs/flink/helpers/Utils.java b/samoa-flink/src/main/java/com/yahoo/labs/flink/com/yahoo/labs/flink/helpers/Utils.java
new file mode 100644
index 0000000..fe1b960
--- /dev/null
+++ b/samoa-flink/src/main/java/com/yahoo/labs/flink/com/yahoo/labs/flink/helpers/Utils.java
@@ -0,0 +1,69 @@
+package com.yahoo.labs.flink.com.yahoo.labs.flink.helpers;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed 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.
+ * #L%
+ */
+
+
+
+import com.yahoo.labs.flink.topology.impl.SamoaType;
+import com.yahoo.labs.samoa.core.ContentEvent;
+import com.yahoo.labs.samoa.utils.PartitioningScheme;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.streaming.api.datastream.DataStream;
+
+import java.util.List;
+
+import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.STRING_TYPE_INFO;
+
+public class Utils {
+
+	public static TypeInformation<SamoaType> tempTypeInfo = new TupleTypeInfo(SamoaType.class, STRING_TYPE_INFO, TypeExtractor.getForClass(ContentEvent.class), STRING_TYPE_INFO);
+
+	public static DataStream subscribe(DataStream<SamoaType> stream, PartitioningScheme partitioning) {
+		switch (partitioning) {
+			case BROADCAST:
+				return stream.broadcast();
+			case GROUP_BY_KEY:
+				return stream.groupBy(new KeySelector<SamoaType, String>() {
+					@Override
+					public String getKey(SamoaType samoaType) throws Exception {
+						return samoaType.f0;
+					}
+				});
+			case SHUFFLE:
+			default:
+				return stream.shuffle();
+		}
+	}
+
+	public static FilterFunction<SamoaType> getFilter(final String streamID) {
+		return new FilterFunction<SamoaType>() {
+			@Override
+			public boolean filter(SamoaType o) throws Exception {
+				return o.f2.equals(streamID);
+			}
+		};
+	}
+
+}
diff --git a/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkComponent.java b/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkComponent.java
new file mode 100644
index 0000000..70a7838
--- /dev/null
+++ b/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkComponent.java
@@ -0,0 +1,68 @@
+package com.yahoo.labs.flink.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed 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.
+ * #L%
+ */
+
+
+
+import org.apache.flink.streaming.api.datastream.DataStream;
+
+/**
+ * Common interface of FlinkEntranceProcessingItem and FlinkProcessingItem
+ */
+public interface FlinkComponent {
+
+	/**
+	 * An initiation of the node. It should create the right invokables and apply the appropriate
+	 * stream transformations
+	 */
+	public void initialise();
+
+	/**
+	 * This check is needed in order to determine whether all requirements for a Flink Component 
+	 * (DataStream) are satisfied in order to initialise it. This is necessary in this integration
+	 * since Flink Streaming applies eager datastream generation based on transformations.
+	 * 
+	 * @return 
+	 */
+	public boolean canBeInitialised();
+
+	/**
+	 * 
+	 * @return
+	 */
+	public boolean isInitialised();
+
+	/**
+	 * The wrapped Flink DataStream generated by this Flink component. Mind that the component 
+	 * should first be initialised in order to have a generated DataStream
+	 * 
+	 * @return
+	 */
+	public DataStream<SamoaType> getOutStream();
+
+	/**
+	 * A unique component id
+	 * 
+	 * @return
+	 */
+	public int getComponentId();
+
+}
diff --git a/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkComponentFactory.java b/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkComponentFactory.java
new file mode 100644
index 0000000..fca0c1a
--- /dev/null
+++ b/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkComponentFactory.java
@@ -0,0 +1,66 @@
+package com.yahoo.labs.flink.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed 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.
+ * #L%
+ */
+
+
+import com.yahoo.labs.samoa.core.EntranceProcessor;
+import com.yahoo.labs.samoa.core.Processor;
+import com.yahoo.labs.samoa.topology.*;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+/**
+ * An implementation of SAMOA's ComponentFactory for Apache Flink
+ */
+public class FlinkComponentFactory implements ComponentFactory {
+
+	private StreamExecutionEnvironment env;
+
+	public FlinkComponentFactory(StreamExecutionEnvironment env) {
+		this.env = env;
+	}
+
+	@Override
+	public ProcessingItem createPi(Processor processor) {
+		return new FlinkProcessingItem(env, processor);
+	}
+
+	@Override
+	public ProcessingItem createPi(Processor processor, int parallelism) {
+		return new FlinkProcessingItem(env, processor, parallelism);
+	}
+
+	@Override
+	public EntranceProcessingItem createEntrancePi(EntranceProcessor entranceProcessor) {
+		return new FlinkEntranceProcessingItem(env, entranceProcessor);
+	}
+
+	@Override
+	public Stream createStream(IProcessingItem sourcePi) {
+		if (sourcePi instanceof FlinkProcessingItem)
+			return ((FlinkProcessingItem) sourcePi).createStream();
+		else return new FlinkStream((FlinkComponent) sourcePi);
+	}
+
+	@Override
+	public Topology createTopology(String topologyName) {
+		return new FlinkTopology(topologyName, env);
+	}
+}
diff --git a/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkEntranceProcessingItem.java b/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkEntranceProcessingItem.java
new file mode 100644
index 0000000..5dca509
--- /dev/null
+++ b/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkEntranceProcessingItem.java
@@ -0,0 +1,101 @@
+package com.yahoo.labs.flink.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed 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.
+ * #L%
+ */
+
+
+import com.yahoo.labs.flink.com.yahoo.labs.flink.helpers.Utils;
+import com.yahoo.labs.samoa.core.EntranceProcessor;
+import com.yahoo.labs.samoa.topology.AbstractEntranceProcessingItem;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.function.source.RichSourceFunction;
+import org.apache.flink.util.Collector;
+
+import java.io.Serializable;
+
+public class FlinkEntranceProcessingItem extends AbstractEntranceProcessingItem
+		implements FlinkComponent, Serializable {
+
+	private transient StreamExecutionEnvironment env;
+	private transient DataStream outStream;
+
+
+	public FlinkEntranceProcessingItem(StreamExecutionEnvironment env, EntranceProcessor proc) {
+		super(proc);
+		this.env = env;
+	}
+
+	@Override
+	public void initialise() {
+		final EntranceProcessor proc = getProcessor();
+		final String streamId = getOutputStream().getStreamId();
+		final int compID = getComponentId();
+
+		
+		outStream = env.addSource(new RichSourceFunction<SamoaType>() {
+			volatile boolean canceled;
+			EntranceProcessor entrProc = proc;
+			String id = streamId;
+
+			@Override
+			public void open(Configuration parameters) throws Exception {
+				super.open(parameters);
+				entrProc.onCreate(compID);
+			}
+
+			@Override
+			public void run(Collector<SamoaType> collector) throws Exception {
+				while (!canceled && entrProc.hasNext()) {
+					collector.collect(SamoaType.of(entrProc.nextEvent(), id));
+				}
+			}
+
+			@Override
+			public void cancel() {
+				canceled = true;
+			}
+		},Utils.tempTypeInfo);
+
+		((FlinkStream) getOutputStream()).initialise();
+	}
+
+
+	@Override
+	public boolean canBeInitialised() {
+		return true;
+	}
+
+	@Override
+	public boolean isInitialised() {
+		return outStream != null;
+	}
+
+	@Override
+	public int getComponentId() {
+		return -1; // dummy number shows that it comes from an Entrance PI
+	}
+
+	@Override
+	public DataStream getOutStream() {
+		return outStream;
+	}
+}
diff --git a/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkProcessingItem.java b/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkProcessingItem.java
new file mode 100644
index 0000000..f92182e
--- /dev/null
+++ b/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkProcessingItem.java
@@ -0,0 +1,248 @@
+package com.yahoo.labs.flink.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed 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.
+ * #L%
+ */
+
+
+import com.google.common.collect.Lists;
+import com.yahoo.labs.flink.com.yahoo.labs.flink.helpers.Utils;
+import com.yahoo.labs.samoa.core.ContentEvent;
+import com.yahoo.labs.samoa.core.Processor;
+import com.yahoo.labs.samoa.topology.ProcessingItem;
+import com.yahoo.labs.samoa.topology.Stream;
+import com.yahoo.labs.samoa.utils.PartitioningScheme;
+import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.invokable.StreamInvokable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+
+public class FlinkProcessingItem extends StreamInvokable<SamoaType, SamoaType> implements ProcessingItem, FlinkComponent, Serializable {
+
+	private static final Logger logger = LoggerFactory.getLogger(FlinkProcessingItem.class);
+	public static final int MAX_WAIT_TIME_MILLIS = 10000;
+
+	private final Processor processor;
+	private final transient StreamExecutionEnvironment env;
+	private final SamoaDelegateFunction fun;
+	private transient DataStream<SamoaType> inStream;
+	private transient DataStream<SamoaType> outStream;
+	private transient List<FlinkStream> outputStreams = Lists.newArrayList();
+	private transient List<Tuple3<FlinkStream, PartitioningScheme, Integer>> inputStreams = Lists.newArrayList();
+	private int parallelism;
+	private static int numberOfPIs = 0;
+	private int piID;
+	private List<Integer> circleId; //check if we can refactor this
+	private boolean onIteration;
+	//private int circleId; //check if we can refactor this
+
+	public FlinkProcessingItem(StreamExecutionEnvironment env, Processor proc) {
+		this(env, proc, 1);
+	}
+
+	public FlinkProcessingItem(StreamExecutionEnvironment env, Processor proc, int parallelism) {
+		this(env, new SamoaDelegateFunction(proc), proc, parallelism);
+	}
+
+	public FlinkProcessingItem(StreamExecutionEnvironment env, SamoaDelegateFunction fun, Processor proc, int parallelism) {
+		super(fun);
+		this.env = env;
+		this.fun = fun;
+		this.processor = proc;
+		this.parallelism = parallelism;
+		this.piID = numberOfPIs++;
+		this.circleId = new ArrayList<Integer>() {
+		}; // if size equals 0, then it is part of no circle
+	}
+
+	public Stream createStream() {
+		FlinkStream generatedStream = new FlinkStream(this);
+		outputStreams.add(generatedStream);
+		return generatedStream;
+	}
+
+	public void putToStream(ContentEvent data, Stream targetStream) {
+		collector.collect(SamoaType.of(data, targetStream.getStreamId()));
+	}
+
+	@Override
+	public void open(Configuration parameters) throws Exception {
+		super.open(parameters);
+		this.processor.onCreate(getComponentId());
+	}
+
+	@Override
+	public void initialise() {
+		for (Tuple3<FlinkStream, PartitioningScheme, Integer> inputStream : inputStreams) {
+			if (inputStream.f0.isInitialised()) { //if input stream is initialised
+				try {
+					DataStream toBeMerged = Utils.subscribe(inputStream.f0.getOutStream(), inputStream.f1);
+					if (inStream == null) {
+						inStream = toBeMerged;
+					} else {
+						inStream = inStream.merge(toBeMerged);
+					}
+				} catch (RuntimeException e) {
+					e.printStackTrace();
+					System.exit(1);
+				}
+			}
+		}
+
+		if (onIteration) {
+			inStream = inStream.iterate(MAX_WAIT_TIME_MILLIS);
+		}
+		outStream = inStream.transform("samoaProcessor", Utils.tempTypeInfo, this).setParallelism(parallelism);
+	}
+
+	public void initialiseStreams() {
+		for (FlinkStream stream : this.getOutputStreams()) {
+			stream.initialise();
+		}
+	}
+
+	@Override
+	public boolean canBeInitialised() {
+		for (Tuple3<FlinkStream, PartitioningScheme, Integer> inputStream : inputStreams) {
+			if (!inputStream.f0.isInitialised()) return false;
+		}
+		return true;
+	}
+
+	@Override
+	public boolean isInitialised() {
+		return outStream != null;
+	}
+
+	@Override
+	public Processor getProcessor() {
+		return processor;
+	}
+
+	@Override
+	public void invoke() throws Exception {
+		while (readNext() != null) {
+			SamoaType t = nextObject;
+			fun.processEvent(t.f1);
+		}
+	}
+
+	@Override
+	public ProcessingItem connectInputShuffleStream(Stream inputStream) {
+		inputStreams.add(new Tuple3<>((FlinkStream) inputStream, PartitioningScheme.SHUFFLE, ((FlinkStream) inputStream).getSourcePiId()));
+		return this;
+	}
+
+	@Override
+	public ProcessingItem connectInputKeyStream(Stream inputStream) {
+		inputStreams.add(new Tuple3<>((FlinkStream) inputStream, PartitioningScheme.GROUP_BY_KEY, ((FlinkStream) inputStream).getSourcePiId()));
+		return this;
+	}
+
+	@Override
+	public ProcessingItem connectInputAllStream(Stream inputStream) {
+		inputStreams.add(new Tuple3<>((FlinkStream) inputStream, PartitioningScheme.BROADCAST, ((FlinkStream) inputStream).getSourcePiId()));
+		return this;
+	}
+
+	@Override
+	public int getParallelism() {
+		return parallelism;
+	}
+
+	public void setParallelism(int parallelism) {
+		this.parallelism = parallelism;
+	}
+
+	public List<FlinkStream> getOutputStreams() {
+		return outputStreams;
+	}
+
+	public DataStream<SamoaType> getOutStream() {
+		return this.outStream;
+	}
+
+	public void setOutStream(DataStream outStream) {
+		this.outStream = outStream;
+	}
+
+	@Override
+	public int getComponentId() {
+		return piID;
+	}
+
+	public boolean isPartOfCircle() {
+		return this.circleId.size() > 0;
+	}
+
+	public List<Integer> getCircleIds() {
+		return circleId;
+	}
+
+	public void addPItoLoop(int piId) {
+		this.circleId.add(piId);
+	}
+
+	public DataStream<SamoaType> getInStream() {
+		return inStream;
+	}
+
+	public List<Tuple3<FlinkStream, PartitioningScheme, Integer>> getInputStreams() {
+		return inputStreams;
+	}
+
+	public void setOnIteration(boolean onIteration) {
+		this.onIteration = onIteration;
+	}
+
+	public boolean isOnIteration() {
+		return onIteration;
+	}
+
+	static class SamoaDelegateFunction implements Function, Serializable {
+		private final Processor proc;
+
+		SamoaDelegateFunction(Processor proc) {
+			this.proc = proc;
+		}
+
+		public void processEvent(ContentEvent event) {
+			proc.process(event);
+		}
+	}
+
+	public FlinkStream getInputStreamBySourceID(int sourceID) {
+		for (Tuple3<FlinkStream, PartitioningScheme, Integer> fstreams : inputStreams) {
+			if (fstreams.f2 == sourceID) {
+				return fstreams.f0;
+			}
+		}
+		return null;
+	}
+
+}
diff --git a/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkStream.java b/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkStream.java
new file mode 100644
index 0000000..c5cb0ed
--- /dev/null
+++ b/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkStream.java
@@ -0,0 +1,94 @@
+package com.yahoo.labs.flink.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed 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.
+ * #L%
+ */
+
+
+
+import com.yahoo.labs.flink.com.yahoo.labs.flink.helpers.Utils;
+import com.yahoo.labs.samoa.core.ContentEvent;
+import com.yahoo.labs.samoa.topology.AbstractStream;
+import org.apache.flink.streaming.api.datastream.DataStream;
+
+import java.io.Serializable;
+
+
+/**
+ * A stream for SAMOA based on Apache Flink's DataStream
+ */
+public class FlinkStream extends AbstractStream implements FlinkComponent, Serializable {
+
+	private static int outputCounter = 0;
+	private FlinkComponent procItem;
+	private transient DataStream<SamoaType> dataStream;
+	private int sourcePiId;
+	private String flinkStreamId;
+
+	public FlinkStream(FlinkComponent sourcePi) {
+		this.procItem = sourcePi;
+		this.sourcePiId = sourcePi.getComponentId();
+		setStreamId("stream-" + Integer.toString(outputCounter));
+		flinkStreamId = "stream-" + Integer.toString(outputCounter);
+		outputCounter++;
+	}
+
+	@Override
+	public void initialise() {
+		if (procItem instanceof FlinkProcessingItem) {
+			dataStream = procItem.getOutStream().filter(Utils.getFilter(getStreamId()))
+			.setParallelism(((FlinkProcessingItem) procItem).getParallelism());
+		} else
+			dataStream = procItem.getOutStream();
+	}
+
+	@Override
+	public boolean canBeInitialised() {
+		return procItem.isInitialised();
+	}
+
+	@Override
+	public boolean isInitialised() {
+		return dataStream != null;
+	}
+
+	@Override
+	public DataStream getOutStream() {
+		return dataStream;
+	}
+
+	@Override
+	public void put(ContentEvent event) {
+		((FlinkProcessingItem) procItem).putToStream(event, this);
+	}
+
+	@Override
+	public int getComponentId() {
+		return -1; //dummy number shows that it comes from a Stream
+	}
+
+	public int getSourcePiId() {
+		return sourcePiId;
+	}
+
+	@Override
+	public String getStreamId() {
+		return flinkStreamId;
+	}
+}
diff --git a/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkTopology.java b/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkTopology.java
new file mode 100644
index 0000000..f04d792
--- /dev/null
+++ b/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/FlinkTopology.java
@@ -0,0 +1,185 @@
+package com.yahoo.labs.flink.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed 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.
+ * #L%
+ */
+
+
+
+import com.google.common.base.Predicate;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.yahoo.labs.flink.com.yahoo.labs.flink.helpers.CircleDetection;
+import com.yahoo.labs.flink.com.yahoo.labs.flink.helpers.Utils;
+import com.yahoo.labs.samoa.topology.AbstractTopology;
+import com.yahoo.labs.samoa.topology.EntranceProcessingItem;
+import com.yahoo.labs.samoa.utils.PartitioningScheme;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.streaming.api.datastream.IterativeDataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A SAMOA topology on Apache Flink
+ * 
+ * A Samoa-Flink Streaming Topology is DAG of ProcessingItems encapsulated within custom operators.
+ * Streams are tagged and filtered in each operator's output so they can be routed to the right 
+ * operator respectively. Building a Flink topology from a Samoa task involves invoking all these
+ * stream transformations and finally, marking and initiating loops in the graph. We have to do that
+ * since Flink only allows explicit loops in the topology started with 'iterate()' and closed with
+ * 'closeWith()'. Thus, when we build a flink topology we have to do it incrementally from the 
+ * sources, mark loops and initialize them with explicit iterations.
+ * 
+ */
+public class FlinkTopology extends AbstractTopology {
+
+	private static final Logger logger = LoggerFactory.getLogger(FlinkTopology.class);
+	public static StreamExecutionEnvironment env;
+	public List<List<FlinkProcessingItem>> topologyLoops = new ArrayList<>();
+	public  List<Integer> backEdges = new ArrayList<Integer>();
+
+	public FlinkTopology(String name, StreamExecutionEnvironment env) {
+		super(name);
+		this.env = env;
+	}
+
+	public StreamExecutionEnvironment getEnvironment() {
+		return env;
+	}
+	
+	public void build() {
+		markCircles();
+		for (EntranceProcessingItem src : getEntranceProcessingItems()) {
+			((FlinkEntranceProcessingItem) src).initialise();
+		}
+		initComponents(ImmutableList.copyOf(Iterables.filter(getProcessingItems(), FlinkProcessingItem.class)));
+	}
+
+	private void initComponents(ImmutableList<FlinkProcessingItem> flinkComponents) {
+		if (flinkComponents.isEmpty()) return;
+
+		for (FlinkProcessingItem comp : flinkComponents) {
+			if (comp.canBeInitialised() && !comp.isInitialised() && !comp.isPartOfCircle()) {
+				comp.initialise();
+				comp.initialiseStreams();
+
+			}//if component is part of one or more circle
+			else if (comp.isPartOfCircle() && !comp.isInitialised()) {
+				for (Integer circle : comp.getCircleIds()) {
+					//check if circle can be initialized
+					if (checkCircleReady(circle)) {
+						logger.debug("Circle: " + circle + " can be initialised");
+						initialiseCircle(circle);
+					} else {
+						logger.debug("Circle cannot be initialised");
+					}
+				}
+			}
+
+		}
+		initComponents(ImmutableList.copyOf(Iterables.filter(flinkComponents, new Predicate<FlinkProcessingItem>() {
+			@Override
+			public boolean apply(FlinkProcessingItem flinkComponent) {
+				return !flinkComponent.isInitialised();
+			}
+		})));
+	}
+
+	private void markCircles(){
+		List<FlinkProcessingItem> pis = Lists.newArrayList(Iterables.filter(getProcessingItems(), FlinkProcessingItem.class));
+		List<Integer>[] graph = new List[pis.size()];
+		FlinkProcessingItem[] processingItems = new FlinkProcessingItem[pis.size()];
+
+
+		for (int i=0;i<pis.size();i++) {
+			graph[i] = new ArrayList<Integer>();
+		}
+		//construct the graph of the topology for the Processing Items (No entrance pi is included)
+		for (FlinkProcessingItem pi: pis) {
+			processingItems[pi.getComponentId()] = pi;
+			for (Tuple3<FlinkStream, PartitioningScheme, Integer> is : pi.getInputStreams()) {
+				if (is.f2 != -1) graph[is.f2].add(pi.getComponentId());
+			}
+		}
+		for (int g=0;g<graph.length;g++)
+			logger.debug(graph[g].toString());
+
+		CircleDetection detCircles = new CircleDetection();
+		List<List<Integer>> circles = detCircles.getCircles(graph);
+
+		//update PIs, regarding being part of a circle.
+		for (List<Integer> c : circles){
+			List<FlinkProcessingItem> circle = new ArrayList<>();
+			for (Integer it : c){
+				circle.add(processingItems[it]);
+				processingItems[it].addPItoLoop(topologyLoops.size());
+			}
+			topologyLoops.add(circle);
+			backEdges.add(circle.get(0).getComponentId());
+		}
+		logger.debug("Circles detected in the topology: " + circles);
+	}
+	
+
+	private boolean checkCircleReady(int circleId) {
+
+		List<Integer> circleIds = new ArrayList<>();
+
+		for (FlinkProcessingItem pi : topologyLoops.get(circleId)) {
+			circleIds.add(pi.getComponentId());
+		}
+		//check that all incoming to the circle streams are initialised
+		for (FlinkProcessingItem procItem : topologyLoops.get(circleId)) {
+			for (Tuple3<FlinkStream, PartitioningScheme, Integer> inputStream : procItem.getInputStreams()) {
+				//if a inputStream is not initialized AND source of inputStream is not in the circle or a tail of other circle
+				if ((!inputStream.f0.isInitialised()) && (!circleIds.contains(inputStream.f2)) && (!backEdges.contains(inputStream.f2)))
+					return false;
+			}
+		}
+		return true;
+	}
+
+	private void initialiseCircle(int circleId) {
+		//get the head and tail of circle
+		FlinkProcessingItem tail = topologyLoops.get(circleId).get(0);
+		FlinkProcessingItem head = topologyLoops.get(circleId).get(topologyLoops.get(circleId).size() - 1);
+
+		//initialise source stream of the iteration, so as to use it for the iteration starting point
+		if (!head.isInitialised()) {
+			head.setOnIteration(true);
+			head.initialise();
+			head.initialiseStreams();
+		}
+
+		//initialise all nodes after head
+		for (int node = topologyLoops.get(circleId).size() - 2; node >= 0; node--) {
+			topologyLoops.get(circleId).get(node).initialise();
+			topologyLoops.get(circleId).get(node).initialiseStreams();
+		}
+
+		((IterativeDataStream) head.getInStream()).closeWith(head.getInputStreamBySourceID(tail.getComponentId()).getOutStream());
+	}
+
+
+}
diff --git a/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/SamoaType.java b/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/SamoaType.java
new file mode 100644
index 0000000..16d050a
--- /dev/null
+++ b/samoa-flink/src/main/java/com/yahoo/labs/flink/topology/impl/SamoaType.java
@@ -0,0 +1,42 @@
+package com.yahoo.labs.flink.topology.impl;
+
+/*
+ * #%L
+ * SAMOA
+ * %%
+ * Copyright (C) 2014 - 2015 Apache Software Foundation
+ * %%
+ * Licensed 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.
+ * #L%
+ */
+
+
+
+
+import com.yahoo.labs.samoa.core.ContentEvent;
+import org.apache.flink.api.java.tuple.Tuple3;
+
+public class SamoaType extends Tuple3<String, ContentEvent, String> {
+	public SamoaType() {
+		super();
+	}
+
+	private SamoaType(String key, ContentEvent event, String streamId) {
+		super(key, event, streamId);
+	}
+
+	public static SamoaType of(ContentEvent event, String streamId) {
+		String key = event.getKey() == null ? "none" : event.getKey();
+		return new SamoaType(key, event, streamId);
+	}
+}
\ No newline at end of file