examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TopWikipediaSessions.java

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.apex.malhar.stream.sample.complete;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Random;
import javax.annotation.Nullable;

import org.joda.time.Duration;

import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.util.KeyValPair;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.Window;
import org.apache.apex.malhar.lib.window.WindowOption;
import org.apache.apex.malhar.lib.window.accumulation.TopN;
import org.apache.apex.malhar.stream.api.ApexStream;
import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
import org.apache.apex.malhar.stream.api.WindowedStream;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
import org.apache.hadoop.conf.Configuration;


import com.datatorrent.api.Context;
import com.datatorrent.api.DAG;
import com.datatorrent.api.DefaultInputPort;
import com.datatorrent.api.DefaultOutputPort;
import com.datatorrent.api.InputOperator;
import com.datatorrent.api.StreamingApplication;
import com.datatorrent.api.annotation.ApplicationAnnotation;
import com.datatorrent.common.util.BaseOperator;

import static org.apache.apex.malhar.stream.api.Option.Options.name;

/**
 * Beam's TopWikipediaSessions Example.
 *
 * @since 3.5.0
 */
@ApplicationAnnotation(name = "TopWikipediaSessions")
public class TopWikipediaSessions implements StreamingApplication
{
  /**
   * A generator that outputs a stream of combinations of some users and some randomly generated edit time.
   */
  public static class SessionGen extends BaseOperator implements InputOperator
  {
    private String[] names = new String[]{"user1", "user2", "user3", "user4"};
    public transient DefaultOutputPort<KeyValPair<String, Long>> output = new DefaultOutputPort<>();

    private static final Duration RAND_RANGE = Duration.standardDays(365);
    private Long minTimestamp;
    private long sleepTime;
    private static int tupleCount = 0;

    public static int getTupleCount()
    {
      return tupleCount;
    }

    private String randomName(String[] names)
    {
      int index = new Random().nextInt(names.length);
      return names[index];
    }

    @Override
    public void setup(Context.OperatorContext context)
    {
      super.setup(context);
      tupleCount = 0;
      minTimestamp = System.currentTimeMillis();
      sleepTime = context.getValue(Context.OperatorContext.SPIN_MILLIS);
    }

    @Override
    public void emitTuples()
    {
      long randMillis = (long)(Math.random() * RAND_RANGE.getMillis());
      long randomTimestamp = minTimestamp + randMillis;
      output.emit(new KeyValPair<String, Long>(randomName(names), randomTimestamp));
      tupleCount++;
      try {
        Thread.sleep(sleepTime);
      } catch (InterruptedException e) {
        // Ignore it.
      }
    }
  }

  public static class Collector extends BaseOperator
  {
    private final int resultSize = 5;
    private static List<List<TempWrapper>> result = new ArrayList<>();

    public static List<List<TempWrapper>> getResult()
    {
      return result;
    }

    public final transient DefaultInputPort<Tuple.WindowedTuple<List<TempWrapper>>> input = new DefaultInputPort<Tuple.WindowedTuple<List<TempWrapper>>>()
    {
      @Override
      public void process(Tuple.WindowedTuple<List<TempWrapper>> tuple)
      {
        if (result.size() == resultSize) {
          result.remove(0);
        }
        result.add(tuple.getValue());
      }
    };
  }


  /**
   * Convert the upstream (user, time) combination to a timestamped tuple of user.
   */
  static class ExtractUserAndTimestamp implements Function.MapFunction<KeyValPair<String, Long>, Tuple.TimestampedTuple<String>>
  {
    @Override
    public Tuple.TimestampedTuple<String> f(KeyValPair<String, Long> input)
    {
      long timestamp = input.getValue();
      String userName = input.getKey();

      // Sets the implicit timestamp field to be used in windowing.
      return new Tuple.TimestampedTuple<>(timestamp, userName);

    }
  }

  /**
   * Computes the number of edits in each user session.  A session is defined as
   * a string of edits where each is separated from the next by less than an hour.
   */
  static class ComputeSessions
      extends CompositeStreamTransform<ApexStream<Tuple.TimestampedTuple<String>>, WindowedStream<Tuple.WindowedTuple<KeyValPair<String, Long>>>>
  {
    @Override
    public WindowedStream<Tuple.WindowedTuple<KeyValPair<String, Long>>> compose(ApexStream<Tuple.TimestampedTuple<String>> inputStream)
    {
      return inputStream

        // Chuck the stream into session windows.
        .window(new WindowOption.SessionWindows(Duration.standardHours(1)), new TriggerOption().accumulatingFiredPanes().withEarlyFiringsAtEvery(1))

        // Count the number of edits for a user within one session.
        .countByKey(new Function.ToKeyValue<Tuple.TimestampedTuple<String>, String, Long>()
        {
          @Override
          public Tuple.TimestampedTuple<KeyValPair<String, Long>> f(Tuple.TimestampedTuple<String> input)
          {
            return new Tuple.TimestampedTuple<KeyValPair<String, Long>>(input.getTimestamp(), new KeyValPair<String, Long>(input.getValue(), 1L));
          }
        }, name("ComputeSessions"));
    }
  }

  /**
   * A comparator class used for comparing two TempWrapper objects.
   */
  public static class Comp implements Comparator<TempWrapper>
  {
    @Override
    public int compare(TempWrapper o1, TempWrapper o2)
    {
      return Long.compare(o1.getValue().getValue(), o2.getValue().getValue());
    }
  }

  /**
   * A function to extract timestamp from a TempWrapper object.
   */
  // TODO: Need to revisit and change back to using TimestampedTuple.
  public static class TimestampExtractor implements com.google.common.base.Function<TempWrapper, Long>
  {
    @Override
    public Long apply(@Nullable TempWrapper input)
    {
      return input.getTimestamp();
    }
  }

  /**
   * A temporary wrapper to wrap a KeyValPair and a timestamp together to represent a timestamped tuple, the reason
   * for this is that we cannot resolve a type conflict when calling accumulate(). After the issue resolved, we can
   * remove this class.
   */
  public static class TempWrapper
  {
    private KeyValPair<String, Long> value;
    private Long timestamp;

    public TempWrapper()
    {

    }

    public TempWrapper(KeyValPair<String, Long> value, Long timestamp)
    {
      this.value = value;
      this.timestamp = timestamp;
    }

    @Override
    public String toString()
    {
      return this.value + "  -  " + this.timestamp;
    }

    public Long getTimestamp()
    {
      return timestamp;
    }

    public void setTimestamp(Long timestamp)
    {
      this.timestamp = timestamp;
    }

    public KeyValPair<String, Long> getValue()
    {
      return value;
    }

    public void setValue(KeyValPair<String, Long> value)
    {
      this.value = value;
    }
  }

  /**
   * Computes the longest session ending in each month, in this case we use 30 days to represent every month.
   */
  private static class TopPerMonth
      extends CompositeStreamTransform<ApexStream<Tuple.WindowedTuple<KeyValPair<String, Long>>>, WindowedStream<Tuple.WindowedTuple<List<TempWrapper>>>>
  {

    @Override
    public WindowedStream<Tuple.WindowedTuple<List<TempWrapper>>> compose(ApexStream<Tuple.WindowedTuple<KeyValPair<String, Long>>> inputStream)
    {
      TopN<TempWrapper> topN = new TopN<>();
      topN.setN(10);
      topN.setComparator(new Comp());

      return inputStream

        // Map the input WindowedTuple to a TempWrapper object.
        .map(new Function.MapFunction<Tuple.WindowedTuple<KeyValPair<String, Long>>, TempWrapper>()
        {
          @Override
          public TempWrapper f(Tuple.WindowedTuple<KeyValPair<String, Long>> input)
          {
            Window window = input.getWindows().iterator().next();
            return new TempWrapper(input.getValue(), window.getBeginTimestamp());
          }
        }, name("TempWrapper"))

        // Apply window and trigger option again, this time chuck the stream into fixed time windows.
        .window(new WindowOption.TimeWindows(Duration.standardDays(30)), new TriggerOption().accumulatingFiredPanes().withEarlyFiringsAtEvery(Duration.standardSeconds(5)))

        // Compute the top 10 user-sessions with most number of edits.
        .accumulate(topN, name("TopN")).with("timestampExtractor", new TimestampExtractor());
    }
  }

  /**
   * A map function that combine the user and his/her edit session together to a string and use that string as a key
   * with number of edits in that session as value to create a new key value pair to send to downstream.
   */
  static class SessionsToStringsDoFn implements Function.MapFunction<Tuple.WindowedTuple<KeyValPair<String, Long>>, Tuple.WindowedTuple<KeyValPair<String, Long>>>
  {
    @Override
    public Tuple.WindowedTuple<KeyValPair<String, Long>> f(Tuple.WindowedTuple<KeyValPair<String, Long>> input)
    {
      Window window = input.getWindows().iterator().next();
      return new Tuple.WindowedTuple<KeyValPair<String, Long>>(window, new KeyValPair<String, Long>(
        input.getValue().getKey()  + " : " + window.getBeginTimestamp() + " : " + window.getDurationMillis(),
        input.getValue().getValue()));
    }
  }

  /**
   * A flatmap function that turns the result into readable format.
   */
  static class FormatOutputDoFn implements Function.FlatMapFunction<Tuple.WindowedTuple<List<TempWrapper>>, String>
  {
    @Override
    public Iterable<String> f(Tuple.WindowedTuple<List<TempWrapper>> input)
    {
      ArrayList<String> result = new ArrayList<>();
      for (TempWrapper item : input.getValue()) {
        String session = item.getValue().getKey();
        long count = item.getValue().getValue();
        Window window = input.getWindows().iterator().next();
        result.add(session + " + " + count + " : " + window.getBeginTimestamp());
      }
      return result;
    }
  }

  /**
   * A composite transform that compute the top wikipedia sessions.
   */
  public static class ComputeTopSessions extends CompositeStreamTransform<ApexStream<KeyValPair<String, Long>>, WindowedStream<Tuple.WindowedTuple<List<TempWrapper>>>>
  {
    @Override
    public WindowedStream<Tuple.WindowedTuple<List<TempWrapper>>> compose(ApexStream<KeyValPair<String, Long>> inputStream)
    {
      return inputStream
        .map(new ExtractUserAndTimestamp(), name("ExtractUserAndTimestamp"))
        .addCompositeStreams(new ComputeSessions())
        .map(new SessionsToStringsDoFn(), name("SessionsToStringsDoFn"))
        .addCompositeStreams(new TopPerMonth());
    }
  }

  @Override
  public void populateDAG(DAG dag, Configuration conf)
  {
    SessionGen sg = new SessionGen();
    Collector collector = new Collector();
    StreamFactory.fromInput(sg, sg.output, name("sessionGen"))
      .addCompositeStreams(new ComputeTopSessions())
      .print(name("console"))
      .endWith(collector, collector.input, name("collector")).populateDag(dag);
  }
}