[streaming] Performance test scripts added

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

@@ -30,8 +30,8 @@ public class WordCountCounter extends UserTaskInvokable {
 	private String word = "";
 	private Integer count = 0;
 
-	PerformanceCounter pCounter = new PerformanceCounter("CounterEmitCounter", 1000, 10000);
-	PerformanceTimer pTimer = new PerformanceTimer("CounterEmitTimer", 1000, 10000, true);
+	PerformanceCounter pCounter = new PerformanceCounter("CounterEmitCounter", 1000, 1000);
+	PerformanceTimer pTimer = new PerformanceTimer("CounterEmitTimer", 1000, 1000, true);
 
 	private StreamRecord outRecord = new StreamRecord(new Tuple2<String, Integer>());
 

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

@@ -21,7 +21,7 @@ import eu.stratosphere.streaming.util.PerformanceCounter;
 
 public class WordCountSink extends UserSinkInvokable {
 
-	PerformanceCounter perf = new PerformanceCounter("SinkEmitCounter", 1000, 10000);
+	PerformanceCounter perf = new PerformanceCounter("SinkEmitCounter", 1000, 1000);
 
 	@Override
 	public void invoke(StreamRecord record) throws Exception {

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

@@ -37,7 +37,7 @@ public class WordCountSource extends UserSourceInvokable {
 	@Override
 	public void invoke() throws Exception {
 
-		for (int i = 0; i < 10; i++) {
+		for (int i = 0; i < 2; i++) {
 			try {
 				br = new BufferedReader(new FileReader("/home/strato/strato-dist/resources/hamlet.txt"));
 

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

@@ -25,8 +25,8 @@ public class WordCountSplitter extends UserTaskInvokable {
 
 	private String[] words = new String[] {};
 	private StreamRecord outputRecord = new StreamRecord(new Tuple1<String>());
-	PerformanceCounter pCounter = new PerformanceCounter("SplitterEmitCounter", 1000, 10000);
-	PerformanceTimer pTimer = new PerformanceTimer("SplitterEmitTimer", 1000, 10000, true);
+	PerformanceCounter pCounter = new PerformanceCounter("SplitterEmitCounter", 1000, 1000);
+	PerformanceTimer pTimer = new PerformanceTimer("SplitterEmitTimer", 1000, 1000, true);
 
 	@Override
 	public void invoke(StreamRecord record) throws Exception {

flink-addons/flink-streaming/src/test/resources/Performance/PerformanceTracker.py

@@ -12,21 +12,20 @@ import os
 linestyles = ['_', '-', '--', ':']
 markers=['x','o','^','+']
 def readFiles(csv_dir):
-    dataframes={}
-    machine=[]
+    dataframes=[]
+    
     
     for fname in os.listdir(csv_dir):
         if '.csv' in fname:
-            dataframes[fname.rstrip('.csv')]=pd.read_csv(os.path.join(csv_dir,fname),index_col='Time')
-            machine.append(int(fname.rstrip('.csv')[-1]))
-    return dataframes,machine
+            dataframes.append((fname.rstrip('.csv'),int(fname.rstrip('.csv').split('-')[-1])-1,pd.read_csv(os.path.join(csv_dir,fname),index_col='Time')))
+    return dataframes
     
 def plotCounter(csv_dir, smooth=5):
-    dataframes,machine= readFiles(csv_dir)
+    dataframes= readFiles(csv_dir)
         
     
-    for name in dataframes:
-        df=dataframes[name]
+    for dataframe in dataframes:
+        df=dataframe[2]
         speed=[0]
         values=list(df.ix[:,0])
         for i in range(1,len(values)):
@@ -36,39 +35,44 @@ def plotCounter(csv_dir, smooth=5):
     plt.figure(figsize=(12, 8), dpi=80)
     plt.title('Counter')
     
-    for name in enumerate(dataframes):
-        if len(markers)>machine[name[0]]:
-            m=markers[machine[name[0]]]            
+    for dataframe in dataframes:
+        if len(markers)>dataframe[1]:
+            m=markers[dataframe[1]]            
         else: m='*'  
-
-        dataframes[name[1]].ix[:,0].plot(marker=m,markevery=10,markersize=10)
-    plt.legend(dataframes.keys())
+        dataframe[2].ix[:,0].plot(marker=m,markevery=10,markersize=10)
+    plt.legend([x[0] for x in dataframes])
     
     plt.figure(figsize=(12, 8), dpi=80)
     plt.title('dC/dT')
 
-    for name in enumerate(dataframes):
-        if len(markers)>machine[name[0]]:
-            m=markers[machine[name[0]]]            
-        else: m='*'
-        pd.rolling_mean(dataframes[name[1]].speed,smooth).plot(marker=m,markevery=10,markersize=10)
-    plt.legend(dataframes.keys())
+    for dataframe in dataframes:
+        if len(markers)>dataframe[1]:
+            m=markers[dataframe[1]]            
+        else: m='*'  
+        pd.rolling_mean(dataframe[2].speed,smooth).plot(marker=m,markevery=10,markersize=10)
+    plt.legend([x[0] for x in dataframes])
         
         
 
 def plotTimer(csv_dir,smooth=5,std=50):
-    dataframes,machine= readFiles(csv_dir)
+    dataframes= readFiles(csv_dir)
     
     plt.figure(figsize=(12, 8), dpi=80)
     plt.title('Timer')
     
-    for name in dataframes:
-        pd.rolling_mean(dataframes[name].ix[:,0],smooth).plot()
-    plt.legend(dataframes.keys())
+    for dataframe in dataframes:
+        if len(markers)>dataframe[1]:
+            m=markers[dataframe[1]]            
+        else: m='*'  
+        pd.rolling_mean(dataframe[2].ix[:,0],smooth).plot(marker=m,markevery=10,markersize=10)
+    plt.legend([x[0] for x in dataframes])
     
     plt.figure(figsize=(12, 8), dpi=80)
     plt.title('Standard deviance')
 
-    for name in dataframes:
-        pd.rolling_std(dataframes[name].ix[:,0],std).plot()
-    plt.legend(dataframes.keys())
+    for dataframe in dataframes:
+        if len(markers)>dataframe[1]:
+            m=markers[dataframe[1]]            
+        else: m='*' 
+        pd.rolling_std(dataframe[2].ix[:,0],std).plot(marker=m,markevery=10,markersize=10)
+    plt.legend([x[0] for x in dataframes])

flink-addons/flink-streaming/src/test/resources/Performance/copy-files.sh

+#!/bin/bash
+toDir=$1
+
+if [ -d "${toDir}" ] ; then
+	ssh strato@dell150.ilab.sztaki.hu '
+	for j in {101..142} 144 145;
+	do
+		for i in $(ssh dell$j "ls stratosphere-distrib/log/counter/");
+			do scp strato@dell$j:stratosphere-distrib/log/counter/$i stratosphere-distrib/log/all_tests/counter/$i-$j.csv;
+		done
+		for i in $(ssh dell$j "ls stratosphere-distrib/log/timer/");
+			do scp strato@dell$j:stratosphere-distrib/log/timer/$i stratosphere-distrib/log/all_tests/timer/$i-$j.csv;
+		done
+		for i in $(ls stratosphere-distrib/log/counter/);
+			do cp stratosphere-distrib/log/counter/$i stratosphere-distrib/log/all_tests/counter/$i-150.csv;
+		done
+		for i in $(ls stratosphere-distrib/log/timer/);
+			do cp stratosphere-distrib/log/timer/$i stratosphere-distrib/log/all_tests/timer/$i-150.csv;
+		done
+	done
+	'
+	scp strato@dell150.ilab.sztaki.hu:stratosphere-distrib/log/counter/* $toDir/counter/
+	scp strato@dell150.ilab.sztaki.hu:stratosphere-distrib/log/timer/* $toDir/timer/
+else
+	echo "USAGE:"
+	echo "run <directory>"
+fi
\ No newline at end of file

flink-addons/flink-streaming/src/test/resources/Performance/remove-files.sh

+#!/bin/bash
+ssh strato@dell150.ilab.sztaki.hu '
+for j in {101..142} 144 145;
+do
+	$(ssh dell$j "rm stratosphere-distrib/log/counter/*");
+	$(ssh dell$j "rm stratosphere-distrib/log/timer/*");
+	rm stratosphere-distrib/log/counter/*
+	rm stratosphere-distrib/log/timer/*
+done
+'
\ No newline at end of file