Merge pull request #304 from mwws/streaming

enable checkpoint for fixwindow and wordcount testcase in Flink

Author: wei-mao-intel

conf/01-default-streamingbench.conf

@@ -107,6 +107,8 @@ hibench.streambench.flink.parallelism 20
 
 hibench.streambench.flink.bufferTimeout 5
 
+hibench.streambench.flink.checkpointDuration 1000
+
 #########################################################
 # Storm Config
 #########################################################

src/streambench/common/src/main/java/com/intel/hibench/streambench/common/StreamBenchConfig.java

@@ -86,6 +86,8 @@ public class StreamBenchConfig {
   // ======================================
   public static String FLINK_BUFFERTIMEOUT = "hibench.streambench.flink.bufferTimeout";
 
+  public static String FLINK_CHECKPOINTDURATION = "hibench.streambench.flink.checkpointDuration";
+
   // ======================================
   // Storm Related Conf
   // ======================================

src/streambench/flinkbench/src/main/java/com/intel/flinkbench/RunBench.java

@@ -51,6 +51,7 @@ public static void runAll(String[] args) throws Exception {
         conf.windowDuration = cl.getProperty(StreamBenchConfig.FixWINDOW_DURATION);
         conf.windowSlideStep = cl.getProperty(StreamBenchConfig.FixWINDOW_SLIDESTEP);
 
+        conf.checkpointDuration = Long.parseLong(cl.getProperty(StreamBenchConfig.FLINK_CHECKPOINTDURATION));
         int producerNum = Integer.parseInt(cl.getProperty(StreamBenchConfig.DATAGEN_PRODUCER_NUMBER));
         long recordsPerInterval = Long.parseLong(cl.getProperty(StreamBenchConfig.DATAGEN_RECORDS_PRE_INTERVAL));
         int intervalSpan = Integer.parseInt(cl.getProperty(StreamBenchConfig.DATAGEN_INTERVAL_SPAN));

src/streambench/flinkbench/src/main/java/com/intel/flinkbench/microbench/FixedWindow.java

@@ -18,6 +18,7 @@ public void processStream(final FlinkBenchConfig config) throws Exception{
 
         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
         env.setBufferTimeout(config.bufferTimeout);
+        env.enableCheckpointing(config.checkpointDuration);
 
         createDataStream(config);
         DataStream<Tuple2<String, String>> dataStream = env.addSource(getDataStream());

src/streambench/flinkbench/src/main/java/com/intel/flinkbench/microbench/WordCount.java

@@ -38,6 +38,7 @@ public class WordCount extends StreamBase {
     public void processStream(final FlinkBenchConfig config) throws Exception {
         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
         env.setBufferTimeout(config.bufferTimeout);
+        env.enableCheckpointing(config.checkpointDuration);
         createDataStream(config);
         DataStream<Tuple2<String, String>> dataStream = env.addSource(getDataStream());
         dataStream

src/streambench/flinkbench/src/main/java/com/intel/flinkbench/util/FlinkBenchConfig.java

@@ -39,5 +39,6 @@ public class FlinkBenchConfig implements Serializable{
 
     // Flink related
     public long bufferTimeout;
+    public long checkpointDuration;
 
 }