[streaming] perf util update

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

@@ -59,6 +59,11 @@ public class WordCountSourceSplitter extends UserSourceInvokable {
 				e.printStackTrace();
 			}
 		}
+		
+		Thread.sleep(15000);
+		emit(outRecord);
+		Thread.sleep(15000);
+		emit(outRecord);
 
 	}
 

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/faulttolerance/FaultToleranceUtil.java

@@ -71,7 +71,7 @@ public class FaultToleranceUtil {
 			this.buffer = new AtLeastOnceFaultToleranceBuffer(numberOfChannels, sourceInstanceID);
 		}
 
-		counter = new PerformanceTracker("pc", 1000, 1000, "");
+		counter = new PerformanceTracker("pc", 1000, 1000,14900, "/home/strato/stratosphere-distrib/log/counter/Buffer" + sourceInstanceID);
 
 	}
 

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

@@ -23,7 +23,7 @@ def readFiles(csv_dir):
             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):
+def plotCounter(csv_dir,name='', smooth=5):
     dataframes= readFiles(csv_dir)
         
     
@@ -38,22 +38,47 @@ def plotCounter(csv_dir, smooth=5):
     plt.figure(figsize=(12, 8), dpi=80)
     plt.title('Counter')
     
-    for dataframe in dataframes:
+    if name=='':
+    
+        for dataframe in dataframes:
+            
+            m=markers[dataframe[1]%len(markers)]    
+              
+            dataframe[2].ix[:,0].plot(marker=m,markevery=10,markersize=10)
+        plt.legend([x[0] for x in dataframes])
         
-        m=markers[dataframe[1]%len(markers)]    
-          
-        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')
     
-    plt.figure(figsize=(12, 8), dpi=80)
-    plt.title('dC/dT')
-
-    for dataframe in dataframes:
+        for dataframe in dataframes:
+            
+            m=markers[dataframe[1]%len(markers)]            
+            
+            pd.rolling_mean(dataframe[2].speed,smooth).plot(marker=m,markevery=10,markersize=10)
+        plt.legend([x[0] for x in dataframes])
+    else:
+        df2=[]
+        for dataframe in dataframes:
+            if name in dataframe[0]:
+                 df2.append(dataframe)
         
-        m=markers[dataframe[1]%len(markers)]            
         
-        pd.rolling_mean(dataframe[2].speed,smooth).plot(marker=m,markevery=10,markersize=10)
-    plt.legend([x[0] for x in dataframes])
+        for dataframe in df2:
+            
+            m=markers[dataframe[1]%len(markers)]    
+              
+            dataframe[2].ix[:,0].plot(marker=m,markevery=10,markersize=10)
+        plt.legend([x[0] for x in df2])
+        
+        plt.figure(figsize=(12, 8), dpi=80)
+        plt.title('dC/dT')
+    
+        for dataframe in df2:
+            
+            m=markers[dataframe[1]%len(markers)]            
+            
+            pd.rolling_mean(dataframe[2].speed,smooth).plot(marker=m,markevery=10,markersize=10)
+        plt.legend([x[0] for x in df2])
         
 def plotThroughput(csv_dir,taskname, smooth=5):
     dataframes= readFiles(csv_dir)