[streaming] Implemented getResult() for performance stats in WordCountRemote

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/invokable/UserInvokable.java

@@ -16,5 +16,4 @@
 package eu.stratosphere.streaming.api.invokable;
 
 public interface UserInvokable {
-
 }

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/invokable/UserSinkInvokable.java

@@ -16,8 +16,7 @@
 package eu.stratosphere.streaming.api.invokable;
 
 public abstract class UserSinkInvokable implements RecordInvokable {
-	
-	//TODO: consider moving this to an interface
+	// TODO consider a common interface with StreamInvokableComponents
 	public String getResult() {
 		return "Override getResult() to pass your own results";
 	}

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamcomponent/StreamComponentHelper.java

@@ -33,7 +33,6 @@ import eu.stratosphere.streaming.api.invokable.DefaultSinkInvokable;
 import eu.stratosphere.streaming.api.invokable.DefaultTaskInvokable;
 import eu.stratosphere.streaming.api.invokable.RecordInvokable;
 import eu.stratosphere.streaming.api.invokable.UserSinkInvokable;
-import eu.stratosphere.streaming.api.streamrecord.TupleSizeMismatchException;
 import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
 import eu.stratosphere.streaming.faulttolerance.AckEvent;
 import eu.stratosphere.streaming.faulttolerance.AckEventListener;

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamcomponent/StreamInvokableComponent.java

@@ -22,13 +22,11 @@ import org.apache.commons.logging.LogFactory;
 
 import eu.stratosphere.nephele.io.RecordWriter;
 import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
-import eu.stratosphere.streaming.faulttolerance.FailEvent;
 import eu.stratosphere.streaming.faulttolerance.FaultToleranceUtil;
 
 public abstract class StreamInvokableComponent {
 
-	private static final Log log = LogFactory
-			.getLog(StreamInvokableComponent.class);
+	private static final Log log = LogFactory.getLog(StreamInvokableComponent.class);
 
 	private List<RecordWriter<StreamRecord>> outputs;
 
@@ -36,8 +34,8 @@ public abstract class StreamInvokableComponent {
 	protected String name;
 	private FaultToleranceUtil emittedRecords;
 
-	public final void declareOutputs(List<RecordWriter<StreamRecord>> outputs,
-			String channelID, String name, FaultToleranceUtil emittedRecords) {
+	public final void declareOutputs(List<RecordWriter<StreamRecord>> outputs, String channelID, String name,
+			FaultToleranceUtil emittedRecords) {
 		this.outputs = outputs;
 		this.channelID = channelID;
 		this.emittedRecords = emittedRecords;
@@ -46,27 +44,30 @@ public abstract class StreamInvokableComponent {
 
 	public final void emit(StreamRecord record) {
 		record.setId(channelID);
-		//emittedRecords.addRecord(record);
+		// emittedRecords.addRecord(record);
 		try {
 			for (RecordWriter<StreamRecord> output : outputs) {
 				output.emit(record);
 				log.info("EMITTED: " + record.getId() + " -- " + name);
 			}
 		} catch (Exception e) {
-			//emittedRecords.failRecord(record.getId());
-			log.warn("FAILED: " + record.getId() + " -- " + name + " -- due to "
-					+ e.getClass().getSimpleName());
+			// emittedRecords.failRecord(record.getId());
+			log.warn("FAILED: " + record.getId() + " -- " + name + " -- due to " + e.getClass().getSimpleName());
 		}
 	}
-	
-	//TODO: Add fault tolerance
+
+	// TODO: Add fault tolerance
 	public final void emit(StreamRecord record, int outputChannel) {
 		record.setId(channelID);
-		emittedRecords.addRecord(record,outputChannel);
+		emittedRecords.addRecord(record, outputChannel);
 		try {
 			outputs.get(outputChannel).emit(record);
 		} catch (Exception e) {
 			log.warn("EMIT ERROR: " + e.getClass().getSimpleName() + " -- " + name);
 		}
 	}
+
+	public String getResult() {
+		return "Override getResult() to pass your own results";
+	}
 }

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamcomponent/StreamSource.java

@@ -91,6 +91,8 @@ public class StreamSource extends AbstractInputTask<DummyIS> {
 	public void invoke() throws Exception {
 		log.debug("SOURCE " + name + " invoked with instance id " + sourceInstanceID);
 		userFunction.invoke();
+		// TODO print to file
+		System.out.println(userFunction.getResult());
 	}
 
 }

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamcomponent/StreamTask.java

@@ -86,6 +86,8 @@ public class StreamTask extends AbstractTask {
 	public void invoke() throws Exception {
 		log.debug("TASK " + name + " invoked with instance id " + taskInstanceID);
 		streamTaskHelper.invokeRecords(userFunction, inputs, name);
+		// TODO print to file
+		System.out.println(userFunction.getResult());
 		log.debug("TASK " + name + " invoke finished with instance id " + taskInstanceID);
 	}
 

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/examples/wordcount/WordCountRemote.java

@@ -18,6 +18,8 @@ package eu.stratosphere.streaming.examples.wordcount;
 import java.io.File;
 import java.net.InetSocketAddress;
 import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
 import java.util.Map;
 
 import org.apache.log4j.Level;
@@ -37,21 +39,29 @@ import eu.stratosphere.streaming.api.streamrecord.StreamRecord;
 import eu.stratosphere.streaming.util.LogUtils;
 
 public class WordCountRemote {
+	private final static int recordsEmitted = 100000;
+	private final static int statPerRecords = 10000;
+	private final static int recordsEmittedFromSplitter = 350000; // recordsEmitted * wordsEmittedInARecord
+	private final static int statPerRecordsAfterSplit = recordsEmittedFromSplitter / recordsEmitted * statPerRecords;
 
 	public static class WordCountDebugSource extends UserSourceInvokable {
+		int statCounter = 0;
+		int[] atNumOfRecords = new int[recordsEmitted / statPerRecords + 1];
+		long[] times = new long[recordsEmitted / statPerRecords + 1];
 
 		StreamRecord record = new StreamRecord(new Tuple1<String>());
 
 		@Override
 		public void invoke() throws Exception {
-			long time = System.currentTimeMillis();
-
-			for (int i = 1; i <= 100000; i++) {
-
-				if (i % 10000 == 0) {
-					System.out.println("Source: " + i + "\t time: "
-							+ (System.currentTimeMillis() - time) + " ms");
-					time = System.currentTimeMillis();
+			atNumOfRecords[statCounter] = 0;
+			times[statCounter] = System.currentTimeMillis();
+			statCounter++;
+
+			for (int i = 1; i <= recordsEmitted; i++) {
+				if (i % statPerRecords == 0) {
+					atNumOfRecords[statCounter] = i;
+					times[statCounter] = System.currentTimeMillis();
+					statCounter++;
 				}
 
 				if (i % 2 == 0) {
@@ -62,22 +72,36 @@ public class WordCountRemote {
 				emit(record);
 			}
 		}
+
+		@Override
+		public String getResult() {
+			StringBuilder result = new StringBuilder("");
+			result.append("Source result:\n");
+			for (int i = 0; i < atNumOfRecords.length; i++) {
+				result.append(atNumOfRecords[i] + ";" + times[i] + ";\n");
+			}
+			return result.toString();
+		}
 	}
 
 	public static class WordCountDebugSplitter extends UserTaskInvokable {
+		int statCounter = 0;
+		int[] atNumOfRecords = new int[recordsEmitted / statPerRecords + 1];
+		long[] times = new long[atNumOfRecords.length];
 
 		private String[] words = new String[] {};
 		private int i = 0;
 		private StreamRecord outputRecord = new StreamRecord(new Tuple1<String>());
 		long time = System.currentTimeMillis();
 
+		
 		@Override
 		public void invoke(StreamRecord record) throws Exception {
 			i++;
-			if (i % 10000 == 0) {
-				System.out.println("Splitter: " + i + "\t time: "
-						+ (System.currentTimeMillis() - time) + " ms");
-				time = System.currentTimeMillis();
+			if (i % statPerRecords == 0) {
+				atNumOfRecords[statCounter] = i;
+				times[statCounter] = System.currentTimeMillis();
+				statCounter++;
 			}
 
 			words = record.getString(0).split(" ");
@@ -86,16 +110,28 @@ public class WordCountRemote {
 				emit(outputRecord);
 			}
 		}
+
+		@Override
+		public String getResult() {
+			StringBuilder result = new StringBuilder("");
+			result.append("Splitter result:\n");
+			for (int i = 0; i < atNumOfRecords.length; i++) {
+				result.append(atNumOfRecords[i] + ";" + times[i] + ";\n");
+			}
+			return result.toString();
+		}
 	}
 
 	public static class WordCountDebugCounter extends UserTaskInvokable {
-
+		int statCounter = 0;
+		int[] atNumOfRecords = new int[recordsEmitted / statPerRecords + 1];
+		long[] times = new long[atNumOfRecords.length];
+		
 		private Map<String, Integer> wordCounts = new HashMap<String, Integer>();
 		private String word = "";
 		private Integer count = 0;
 
 		private int i = 0;
-		private long time = System.currentTimeMillis();
 
 		private StreamRecord outRecord = new StreamRecord(new Tuple2<String, Integer>());
 
@@ -103,11 +139,13 @@ public class WordCountRemote {
 		public void invoke(StreamRecord record) throws Exception {
 			word = record.getString(0);
 			i++;
-			if (i % 30000 == 0) {
-				System.out.println("Counter: " + i + "\t time: "
-						+ (System.currentTimeMillis() - time) + " ms");
-				time = System.currentTimeMillis();
+			
+			if (i % statPerRecordsAfterSplit == 0) {
+				atNumOfRecords[statCounter] = i;
+				times[statCounter] = System.currentTimeMillis();
+				statCounter++;
 			}
+			
 			if (wordCounts.containsKey(word)) {
 				count = wordCounts.get(word) + 1;
 				wordCounts.put(word, count);
@@ -121,28 +159,42 @@ public class WordCountRemote {
 
 			emit(outRecord);
 		}
+		
+		@Override
+		public String getResult() {
+			StringBuilder result = new StringBuilder("");
+			result.append("Counter result:\n");
+			for (int i = 0; i < atNumOfRecords.length; i++) {
+				result.append(atNumOfRecords[i] + ";" + times[i] + ";\n");
+			}
+			return result.toString();
+		}
 	}
 
 	public static class WordCountDebugSink extends UserSinkInvokable {
-
 		int nrOfRecords = 0;
-		private long time = System.currentTimeMillis();
-
+		int[] atNumOfRecords = new int[recordsEmitted / statPerRecords + 1];
+		long[] times = new long[atNumOfRecords.length];
+		
 		@Override
 		public void invoke(StreamRecord record) throws Exception {
-			nrOfRecords++;
-			if (nrOfRecords % 30000 == 0) {
-				System.out.println("Sink: " + nrOfRecords + "\t time: "
-						+ (System.currentTimeMillis() - time) + " ms");
-				time = System.currentTimeMillis();
+			if (nrOfRecords % statPerRecordsAfterSplit == 0) {
+				atNumOfRecords[nrOfRecords] = nrOfRecords;
+				times[nrOfRecords] = System.currentTimeMillis();
+				nrOfRecords++;
 			}
 		}
 
 		@Override
 		public String getResult() {
-			return String.valueOf(nrOfRecords);
+			StringBuilder result = new StringBuilder("");
+			result.append("Sink result:\n");
+			for (int i = 0; i < atNumOfRecords.length; i++) {
+				result.append(atNumOfRecords[i] + ";" + times[i] + ";\n");
+			}
+			result.append("RESULT: " + nrOfRecords);
+			return result.toString();
 		}
-
 	}
 
 	private static JobGraph getJobGraph() throws Exception {
@@ -171,8 +223,7 @@ public class WordCountRemote {
 			jG.addJar(new Path(file.getAbsolutePath()));
 
 			Configuration configuration = jG.getJobConfiguration();
-			Client client = new Client(new InetSocketAddress("hadoop00.ilab.sztaki.hu", 6123),
-					configuration);
+			Client client = new Client(new InetSocketAddress("hadoop00.ilab.sztaki.hu", 6123), configuration);
 			client.run(jG, true);
 		} catch (Exception e) {
 			System.out.println(e);