package com.oreilly.learningsparkexamples.java.logs;

import com.google.common.collect.Ordering;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;
import scala.Tuple4;

import java.io.Serializable;
import java.util.Comparator;
import java.util.List;

public class LogAnalyzerWindowed implements Serializable {
  private LogStatistics logStatistics;

  public void processAccessLogs(JavaDStream<ApacheAccessLog> accessLogsDStream) {
    JavaDStream<ApacheAccessLog> windowDStream = accessLogsDStream.window(
        Flags.getInstance().getWindowLength(),
        Flags.getInstance().getSlideInterval());
    JavaDStream<String> ipAddresses = accessLogsDStream.map(
      new Function<ApacheAccessLog, String>() {
        public String call(ApacheAccessLog entry) {
          return entry.getIpAddress();
        }});
    // reduceByWindow
    JavaDStream<Long> requestCountRBW = accessLogsDStream.map(new Function<ApacheAccessLog, Long>() {
        public Long call(ApacheAccessLog entry) {
          return 1L;
        }}).reduceByWindow(new Function2<Long, Long, Long>() {
            public Long call(Long v1, Long v2) {
              return v1+v2;
            }}, new Function2<Long, Long, Long>() {
            public Long call(Long v1, Long v2) {
              return v1-v2;
            }}, Flags.getInstance().getWindowLength(), Flags.getInstance().getSlideInterval());
    requestCountRBW.print();
    // Use countByWindow
    JavaDStream<Long> requestCount = accessLogsDStream.countByWindow(
      Flags.getInstance().getWindowLength(), Flags.getInstance().getSlideInterval());
    JavaPairDStream<String, Long> ipAddressRequestCount = ipAddresses.countByValueAndWindow(
      Flags.getInstance().getWindowLength(), Flags.getInstance().getSlideInterval());
    requestCount.print();
    ipAddressRequestCount.print();

    // use a transform for the response code count
    JavaPairDStream<Integer, Long> responseCodeCountTransform = accessLogsDStream.transformToPair(
      new Function<JavaRDD<ApacheAccessLog>, JavaPairRDD<Integer, Long>>() {
        public JavaPairRDD<Integer, Long> call(JavaRDD<ApacheAccessLog> logs) {
          return Functions.responseCodeCount(logs);
        }
      });
    windowDStream.foreachRDD(new Function<JavaRDD<ApacheAccessLog>, Void>() {
        public Void call(JavaRDD<ApacheAccessLog> accessLogs) {
      Tuple4<Long, Long, Long, Long> contentSizeStats =
          Functions.contentSizeStats(accessLogs);

      List<Tuple2<Integer, Long>> responseCodeToCount =
          Functions.responseCodeCount(accessLogs)
          .take(100);

      JavaPairRDD<String, Long> ipAddressCounts =
          Functions.ipAddressCount(accessLogs);
      List<String> ipAddresses = Functions.filterIPAddress(ipAddressCounts)
          .take(100);

      Object ordering = Ordering.natural();
      Comparator<Long> cmp = (Comparator<Long>)ordering;
      List<Tuple2<String, Long>> topEndpoints =
          Functions.endpointCount(accessLogs)
        .top(10, new Functions.ValueComparator<String, Long>(cmp));

      logStatistics = new LogStatistics(contentSizeStats, responseCodeToCount,
          ipAddresses, topEndpoints);
      return null;
        }});
  }

  public LogStatistics getLogStatistics() {
    return logStatistics;
  }
}

@@ -1730,24 +1730,899 @@
 BW.print();%0A
+    // reducebykeyandwindow%0A    JavaPairDStream%3CString, Long%3E ipAddressPairDStream = accessLogsDStream.mapToPair(%0A      new PairFunction%3CApacheAccessLog, String, Long%3E() %7B%0A        public Tuple2%3CString, Long%3E call(ApacheAccessLog entry) %7B%0A          return new Tuple2(entry.getIpAddress(), 1L);%0A        %7D%7D);%0A    JavaPairDStream%3CString, Long%3E ipAddressesCountDStream = ipAddressPairDStream.reduceByKeyAndWindow(%0A      // Adding elements in the new slice%0A      new Function2%3CLong, Long, Long%3E() %7B%0A        public Long call(Long v1, Long v2) %7B%0A          return v1+v2;%0A        %7D%7D,%0A      // Removing elements from the oldest slice%0A      new Function2%3CLong, Long, Long%3E() %7B%0A        public Long call(Long v1, Long v2) %7B%0A          return v1-v2;%0A        %7D%7D,%0A      Flags.getInstance().getWindowLength(),%0A      Flags.getInstance().getSlideInterval());%0A    ipAddressesCountDStream.print();%0A
     // Use c

add reducebykeyandwindow example for java