Average time between peaks

gistfile1.java

public Double computeAvgTimeBetweenPeak(JavaRDD<long[]> data) {
    // define the maximum using the max function from MLlib
    double[] max = this.summary.max().toArray();

    // keep the timestamp of data point for which the value is greater than 0.9 * max
    // and sort it !
    JavaRDD<Long> filtered_y = data.filter(record -> record[1] > 0.9 * max[1])
                                   .map(record -> record[0])
                                   .sortBy(time -> time, true, 1);

    if (filtered_y.count() > 1) {
      Long firstElement = filtered_y.first();
      Long lastElement = filtered_y.sortBy(time -> time, false, 1).first();

      // compute the delta between each tick
      JavaRDD<Long> firstRDD = filtered_y.filter(record -> record > firstElement);
      JavaRDD<Long> secondRDD = filtered_y.filter(record -> record < lastElement);

      JavaRDD<Vector> product = firstRDD.zip(secondRDD)
                                        .map(pair -> pair._1() - pair._2())
                                         // and keep it if the delta is != 0
                                        .filter(value -> value > 0)
                                        .map(line -> Vectors.dense(line));

      // compute the mean of the delta
      return Statistics.colStats(product.rdd()).mean().toArray()[0];
    }

    return 0.0;
  }