[FLINK] Changes to SystemML

Changes to SystemML in order to support Flink.

pom.xml

@@ -68,6 +68,8 @@
 		<spark.version>1.4.1</spark.version>
 		<scala.version>2.10.5</scala.version>
 		<scala.binary.version>2.10</scala.binary.version>
+		<flink.version>1.0.0</flink.version>
+
 		<!-- OS-specific JVM arguments for running integration tests -->
 		<integrationTestExtraJVMArgs />
 	</properties>
@@ -754,6 +756,35 @@
 
 
 	<dependencies>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-java</artifactId>
+			<version>${flink.version}</version>
+			<scope>provided</scope>
+			<exclusions>
+				<exclusion>
+						<groupId>javax.servlet</groupId>
+						<artifactId>servlet-api</artifactId>
+				</exclusion>
+				<exclusion>
+					<groupId>org.apache.flink</groupId>
+					<artifactId>flink-shaded-hadoop2</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-clients_${scala.binary.version}</artifactId>
+			<version>${flink.version}</version>
+			<scope>provided</scope>
+			<exclusions>
+				<exclusion>
+					<groupId>org.apache.flink</groupId>
+					<artifactId>flink-shaded-hadoop2</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+
 		<dependency>
 			<groupId>org.apache.spark</groupId>
 			<artifactId>spark-core_2.10</artifactId>

src/main/java/org/apache/sysml/api/DMLScript.java

@@ -69,6 +69,7 @@
 import org.apache.sysml.runtime.controlprogram.caching.CacheableData;
 import org.apache.sysml.runtime.controlprogram.context.ExecutionContext;
 import org.apache.sysml.runtime.controlprogram.context.ExecutionContextFactory;
+import org.apache.sysml.runtime.controlprogram.context.FlinkExecutionContext;
 import org.apache.sysml.runtime.controlprogram.context.SparkExecutionContext;
 import org.apache.sysml.runtime.controlprogram.parfor.ProgramConverter;
 import org.apache.sysml.runtime.controlprogram.parfor.stat.InfrastructureAnalyzer;
@@ -93,7 +94,9 @@ public enum RUNTIME_PLATFORM {
 		SINGLE_NODE,    // execute all matrix operations in CP
 		HYBRID,         // execute matrix operations in CP or MR
 		HYBRID_SPARK,   // execute matrix operations in CP or Spark   
-		SPARK			// execute matrix operations in Spark
+		SPARK,			// execute matrix operations in Spark
+		FLINK,
+		HYBRID_FLINK
 	}
 
 	public static RUNTIME_PLATFORM rtplatform = RUNTIME_PLATFORM.HYBRID; //default exec mode
@@ -520,6 +523,10 @@ else if ( platform.equalsIgnoreCase("spark"))
 			lrtplatform = RUNTIME_PLATFORM.SPARK;
 		else if ( platform.equalsIgnoreCase("hybrid_spark"))
 			lrtplatform = RUNTIME_PLATFORM.HYBRID_SPARK;
+        else if ( platform.equalsIgnoreCase("flink"))
+            lrtplatform = RUNTIME_PLATFORM.FLINK;
+		else if ( platform.equalsIgnoreCase("hybrid_flink"))
+			lrtplatform = RUNTIME_PLATFORM.HYBRID_FLINK;
 		else 
 			System.err.println("ERROR: Unknown runtime platform: " + platform);
 
@@ -674,8 +681,11 @@ private static void execute(String dmlScriptStr, String fnameOptConfig, Map<Stri
 
 			//run execute (w/ exception handling to ensure proper shutdown)
 			ec = ExecutionContextFactory.createContext(rtprog);
-			rtprog.execute( ec );  
-
+			rtprog.execute( ec );
+			if (ec instanceof FlinkExecutionContext) {
+				((FlinkExecutionContext) ec).execute();
+
+			}
 		}
 		finally //ensure cleanup/shutdown
 		{