Added change_point code and benchmark

BenchStand/src/main/java/StreamProcessing/BenchmarkChangePoint.java

@@ -0,0 +1,96 @@
+package StreamProcessing;
+
+import akka.actor.Identify;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.shaded.guava18.com.google.common.base.Functions;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import java.io.*;
+import java.util.ArrayList;
+import java.util.List;
+
+public class BenchmarkChangePoint {
+
+    public static void main(String[] args) throws Exception {
+        final Config config;
+        if (args.length > 0) {
+            config = ConfigFactory.parseFile(new File(args[0]));
+        } else {
+            config = ConfigFactory.load("bench.conf");
+        }
+
+        final BenchConfig benchConfig = new BenchConfig(
+                config.getString("benchstand.file_path"),
+                config.getInt("benchstand.word_count"),
+                config.getInt("benchstand.drop_first_n_words"),
+                config.hasPath("remote") ? config.getString("remote.manager_hostname") : null,
+                config.hasPath("remote") ? config.getInt("remote.manager_port") : -1,
+                config.getInt("job.computation_delay"),
+                config.getInt("job.validation_delay"),
+                config.getInt("job.log_each"),
+                config.getString("job.bench_host"),
+                config.getInt("job.source_port"),
+                config.getInt("job.sink_port"));
+
+        final List<Integer> parallelism = config.getIntList("benchstand.parallelism");
+
+        final String log = System.getProperty("user.home") + "/results.txt";
+        try (PrintWriter pw = new PrintWriter(log, "UTF-8")) {
+            pw.println();
+            pw.println("Shuffle on N nodes, then merge on 1 node");
+            benchConfig.mergeOnSingleNode = true;
+
+            for (int p : parallelism) {
+                pw.println("Parallelism " + p);
+                benchConfig.parallelism = p;
+                for (int j = 0; j < 5; j++) {
+                    pw.println(benchmark(benchConfig));
+                    pw.flush();
+                }
+            }
+        }
+    }
+
+    public static BenchResult benchmark(BenchConfig config) throws Exception {
+        final Runnable deployer = () -> {
+            final StreamExecutionEnvironment env;
+            int parallelism = config.parallelism;
+
+            if (config.managerHostname != null) {
+                env = StreamExecutionEnvironment.createRemoteEnvironment(
+                        config.managerHostname, config.managerPort,
+                        parallelism, "BenchStand.jar");
+            } else {
+                env = StreamExecutionEnvironment.createLocalEnvironment(parallelism);
+            }
+
+            final SingleOutputStreamOperator<Stat> words =
+                    env.addSource(new KryoSocketSource(config.benchHost, config.sourcePort))
+                            .keyBy("word")
+                            .map(new WindowAlgoProcessor())
+                            .setParallelism(parallelism);
+            words.map(x -> x)
+                .setParallelism(1)
+                .map(new FusionRule())
+                .addSink(new KryoSocketSink(config.benchHost, config.sinkPort))
+                .setParallelism(1);
+            env.setBufferTimeout(0);
+
+            new Thread(() -> {
+                try {
+                    env.execute();
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+            }, "Flink Job").start();
+        };
+
+        try (BenchStand benchStand = new BenchStand(config, deployer)) {
+            return benchStand.run();
+        }
+    }
+}

BenchStand/src/main/resources/bench.conf

@@ -3,7 +3,7 @@
     "file_path": "TestData/comsn10.txt",
     "word_count": 21932,
     "drop_first_n_words": 1000,
-    "parallelism": [1, 2],
+    "parallelism": [1, 4, 8, 16],
     "steps": [1, 1.5],
     "min_delay_between_words": 1700000,
     "max_delay_between_words": 2100000,

FlinkOperators/src/main/java/StreamProcessing/FusionRule.java

@@ -0,0 +1,42 @@
+package StreamProcessing;
+
+import org.apache.flink.api.common.functions.MapFunction;
+
+import java.util.TreeMap;
+import java.util.function.BinaryOperator;
+
+public class FusionRule implements MapFunction<Stat, Integer> {
+    private final TreeMap<Integer, double[]> actual = new TreeMap<>();
+    private int cnt = 0;
+
+    public static final double[] thresholds = new double[]{0.24, 0.15, 0.9, 0.7};
+
+    @Override
+    public Integer map(Stat stat) throws Exception {
+        actual.put(stat.node, stat.stats);
+        double[] stats = actual.values().stream()
+            .reduce(
+                new double[4],
+                (doubles, doubles2) -> {
+                    double[] ans = new double[4];
+                    for (int i = 0; i < 4; i++) {
+                        ans[i] = doubles[i] + doubles2[i];
+                    }
+                    return ans;
+                }
+            );
+        int flag = 0;
+        for (int i = 0; i < 4; i++) {
+            stats[i] /= actual.values().size();
+            if (stats[i] > thresholds[i]) {
+                flag = 1;
+            }
+        }
+        cnt++;
+        /*if (cnt % 1000 == 0) {
+            System.out.println(String.format("Flag = %d", flag));
+        }*/
+        //return stat.id;
+        return flag;
+    }
+}

FlinkOperators/src/main/java/StreamProcessing/Pass.java

@@ -14,6 +14,7 @@ public Pass(int mean) {
 
     @Override
     public WordWithID map(WordWithID word) {
+        System.out.println(String.format("Id: %d", this.hashCode()));
         LockSupport.parkNanos((long) exp.sample());
         return word;
     }

FlinkOperators/src/main/java/StreamProcessing/Stat.java

@@ -0,0 +1,13 @@
+package StreamProcessing;
+
+public class Stat {
+    public final double[] stats;
+    public int node;
+    public int id;
+
+    Stat(double[] stats, int node, int id) {
+        this.stats = stats;
+        this.node = node;
+        this.id = id;
+    }
+}

FlinkOperators/src/main/java/StreamProcessing/WindowAlgoProcessor.java

@@ -0,0 +1,71 @@
+package StreamProcessing;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.Collector;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+public class WindowAlgoProcessor implements MapFunction<WordWithID, Stat> {
+  private static final int[] WINDOW_SIZES = new int[]{200, 400, 800, 1600};
+  private final ArrayList<ArrayDeque<Double>> referenceWindows = new ArrayList<>();
+  private final ArrayList<ArrayDeque<Double>> slidingWindows = new ArrayList<>();
+  private final double[] ans = new double[4];
+  private int cnt = 0;
+
+  public WindowAlgoProcessor() {
+    System.out.println(String.format("Created node %d", this.hashCode()));
+    for (int i = 0; i < 4; i++) {
+      referenceWindows.add(new ArrayDeque<>());
+      slidingWindows.add(new ArrayDeque<>());
+      ans[i] = 0;
+    }
+  }
+
+  private double kolmogorovSmirnovDistance(Collection<Double> sample1, Collection<Double> sample2) {
+    ArrayList<Tuple2<Double, Double>> kek = sample1.stream()
+            .map(x -> new Tuple2<>(x, -1 / (double) sample1.size()))
+            .collect(Collectors.toCollection(ArrayList::new));
+    kek.addAll(
+            sample2.stream()
+                    .map(x -> new Tuple2<>(x, 1 / (double) sample2.size()))
+                    .collect(Collectors.toCollection(ArrayList::new))
+    );
+    kek.sort(Comparator.comparing(x -> x.f0));
+    double max = 0;
+    double prefix = 0;
+    for (Tuple2<Double, Double> pair: kek) {
+      prefix += pair.f1;
+      max = Math.max(max, Math.abs(prefix));
+    }
+    return max;
+  }
+
+  @Override
+  public Stat map(WordWithID word) {
+    double value = word.word.length();
+    cnt++;
+    for (int i = 0; i < 4; i++) {
+      if (referenceWindows.get(i).size() < WINDOW_SIZES[i]) {
+        referenceWindows.get(i).addLast(value);
+      } else if (slidingWindows.get(i).size() < WINDOW_SIZES[i]) {
+        slidingWindows.get(i).addLast(value);
+      } else {
+        slidingWindows.get(i).removeFirst();
+        slidingWindows.get(i).addLast(value);
+        ans[i] = Math.max(ans[i], kolmogorovSmirnovDistance(referenceWindows.get(i), slidingWindows.get(i)));
+        /*if (cnt % 1000 == 0) {
+          System.out.println(String.format("Node = %d, size = %d, stat = %f"
+                  , this.hashCode()
+                  , WINDOW_SIZES[i]
+                  , ans[i]
+          ));
+        }*/
+      }
+    }
+
+    return new Stat(ans.clone(), this.hashCode(), word.id);
+  }
+}

StatisticalTests/shards_plots.ipynb

@@ -241,7 +241,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.7.0"
+   "version": "3.7.6"
   }
  },
  "nbformat": 4,