content: /**
   * 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.cookbook;
  
  import java.util.Date;
  import java.util.Objects;
  
  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.WindowOption;
  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;
  
  /**
   * This example illustrates the basic concepts behind triggering. It shows how to use different
   * trigger definitions to produce partial (speculative) results before all the data is processed and
   * to control when updated results are produced for late data. The example performs a streaming
   * analysis of the data coming in from PubSub and writes the results to BigQuery. It divides the
   * data into {@link Window windows} to be processed, and demonstrates using various kinds of
   * {@link org.apache.beam.sdk.transforms.windowing.Trigger triggers} to control when the results for
   * each window are emitted.
   *
   * <p> This example uses a portion of real traffic data from San Diego freeways. It contains
   * readings from sensor stations set up along each freeway. Each sensor reading includes a
   * calculation of the 'total flow' across all lanes in that freeway direction.
   *
   * <p> Concepts:
   * <pre>
   *   1. The default triggering behavior
   *   2. Late data with the default trigger
   *   3. How to get speculative estimates
   *   4. Combining late data and speculative estimates
   * </pre>
   *
   * <p> Before running this example, it will be useful to familiarize yourself with Dataflow triggers
   * and understand the concept of 'late data',
   * See:  <a href="https://cloud.google.com/dataflow/model/triggers">
   * https://cloud.google.com/dataflow/model/triggers </a> and
   * <a href="https://cloud.google.com/dataflow/model/windowing#Advanced">
   * https://cloud.google.com/dataflow/model/windowing#Advanced </a>
   *
   * <p> The example pipeline reads data from a Pub/Sub topic. By default, running the example will
   * also run an auxiliary pipeline to inject data from the default {@code --input} file to the
   * {@code --pubsubTopic}. The auxiliary pipeline puts a timestamp on the injected data so that the
   * example pipeline can operate on <i>event time</i> (rather than arrival time). The auxiliary
   * pipeline also randomly simulates late data, by setting the timestamps of some of the data
   * elements to be in the past. You may override the default {@code --input} with the file of your
   * choosing or set {@code --input=""} which will disable the automatic Pub/Sub injection, and allow
   * you to use a separate tool to publish to the given topic.
   *
   * <p> The example is configured to use the default Pub/Sub topic and the default BigQuery table
   * from the example common package (there are no defaults for a general Dataflow pipeline).
   * You can override them by using the {@code --pubsubTopic}, {@code --bigQueryDataset}, and
   * {@code --bigQueryTable} options. If the Pub/Sub topic or the BigQuery table do not exist,
   * the example will try to create them.
   *
   * <p> The pipeline outputs its results to a BigQuery table.
   * Here are some queries you can use to see interesting results:
   * Replace {@code <enter_table_name>} in the query below with the name of the BigQuery table.
   * Replace {@code <enter_window_interval>} in the query below with the window interval.
   *
   * <p> To see the results of the default trigger,
   * Note: When you start up your pipeline, you'll initially see results from 'late' data. Wait after
   * the window duration, until the first pane of non-late data has been emitted, to see more
   * interesting results.
   * {@code SELECT * FROM enter_table_name WHERE triggerType = "default" ORDER BY window DESC}
   *
   * <p> To see the late data i.e. dropped by the default trigger,
   * {@code SELECT * FROM <enter_table_name> WHERE triggerType = "withAllowedLateness" and
   * (timing = "LATE" or timing = "ON_TIME") and freeway = "5" ORDER BY window DESC, processingTime}
   *
   * <p>To see the the difference between accumulation mode and discarding mode,
   * {@code SELECT * FROM <enter_table_name> WHERE (timing = "LATE" or timing = "ON_TIME") AND
   * (triggerType = "withAllowedLateness" or triggerType = "sequential") and freeway = "5" ORDER BY
   * window DESC, processingTime}
   *
   * <p> To see speculative results every minute,
   * {@code SELECT * FROM <enter_table_name> WHERE triggerType = "speculative" and freeway = "5"
   * ORDER BY window DESC, processingTime}
   *
   * <p> To see speculative results every five minutes after the end of the window
   * {@code SELECT * FROM <enter_table_name> WHERE triggerType = "sequential" and timing != "EARLY"
   * and freeway = "5" ORDER BY window DESC, processingTime}
   *
   * <p> To see the first and the last pane for a freeway in a window for all the trigger types,
   * {@code SELECT * FROM <enter_table_name> WHERE (isFirst = true or isLast = true) ORDER BY window}
   *
   * <p> To reduce the number of results for each query we can add additional where clauses.
   * For examples, To see the results of the default trigger,
   * {@code SELECT * FROM <enter_table_name> WHERE triggerType = "default" AND freeway = "5" AND
   * window = "<enter_window_interval>"}
   *
   * <p> The example will try to cancel the pipelines on the signal to terminate the process (CTRL-C)
   * and then exits.
   *
   * @since 3.5.0
   */
  
  public class TriggerExample
  {
    //Numeric value of fixed window duration, in minutes
    public static final int WINDOW_DURATION = 30;
    // Constants used in triggers.
    // Speeding up ONE_MINUTE or FIVE_MINUTES helps you get an early approximation of results.
    // ONE_MINUTE is used only with processing time before the end of the window
    public static final Duration ONE_MINUTE = Duration.standardMinutes(1);
    // FIVE_MINUTES is used only with processing time after the end of the window
    public static final Duration FIVE_MINUTES = Duration.standardMinutes(5);
    // ONE_DAY is used to specify the amount of lateness allowed for the data elements.
    public static final Duration ONE_DAY = Duration.standardDays(1);
  
    /**
     * This transform demonstrates using triggers to control when data is produced for each window
     * Consider an example to understand the results generated by each type of trigger.
     * The example uses "freeway" as the key. Event time is the timestamp associated with the data
     * element and processing time is the time when the data element gets processed in the pipeline.
     * For freeway 5, suppose there are 10 elements in the [10:00:00, 10:30:00) window.
     * Key (freeway) | Value (totalFlow) | event time | processing time
     * 5             | 50                 | 10:00:03   | 10:00:47
     * 5             | 30                 | 10:01:00   | 10:01:03
     * 5             | 30                 | 10:02:00   | 11:07:00
     * 5             | 20                 | 10:04:10   | 10:05:15
     * 5             | 60                 | 10:05:00   | 11:03:00
     * 5             | 20                 | 10:05:01   | 11.07:30
     * 5             | 60                 | 10:15:00   | 10:27:15
     * 5             | 40                 | 10:26:40   | 10:26:43
     * 5             | 60                 | 10:27:20   | 10:27:25
     * 5             | 60                 | 10:29:00   | 11:11:00
     *
     * <p> Dataflow tracks a watermark which records up to what point in event time the data is
     * complete. For the purposes of the example, we'll assume the watermark is approximately 15m
     * behind the current processing time. In practice, the actual value would vary over time based
     * on the systems knowledge of the current PubSub delay and contents of the backlog (data
     * that has not yet been processed).
     *
     * <p> If the watermark is 15m behind, then the window [10:00:00, 10:30:00) (in event time) would
     * close at 10:44:59, when the watermark passes 10:30:00.
     */
    static class CalculateTotalFlow
        extends CompositeStreamTransform<ApexStream<String>, WindowedStream<SampleBean>>
    {
      private int windowDuration;
  
      CalculateTotalFlow(int windowDuration)
      {
        this.windowDuration = windowDuration;
      }
  
      @Override
      public WindowedStream<SampleBean> compose(ApexStream<String> inputStream)
      {
        // Concept #1: The default triggering behavior
        // By default Dataflow uses a trigger which fires when the watermark has passed the end of the
        // window. This would be written {@code Repeatedly.forever(AfterWatermark.pastEndOfWindow())}.
  
        // The system also defaults to dropping late data -- data which arrives after the watermark
        // has passed the event timestamp of the arriving element. This means that the default trigger
        // will only fire once.
  
        // Each pane produced by the default trigger with no allowed lateness will be the first and
        // last pane in the window, and will be ON_TIME.
  
        // The results for the example above with the default trigger and zero allowed lateness
        // would be:
        // Key (freeway) | Value (totalFlow) | numberOfRecords | isFirst | isLast | timing
        // 5             | 260                | 6                 | true    | true   | ON_TIME
  
        // At 11:03:00 (processing time) the system watermark may have advanced to 10:54:00. As a
        // result, when the data record with event time 10:05:00 arrives at 11:03:00, it is considered
        // late, and dropped.
  
        WindowedStream<SampleBean> defaultTriggerResults = inputStream
            .window(new WindowOption.TimeWindows(Duration.standardMinutes(windowDuration)),
            new TriggerOption().discardingFiredPanes())
            .addCompositeStreams(new TotalFlow("default"));
  
        // Concept #2: Late data with the default trigger
        // This uses the same trigger as concept #1, but allows data that is up to ONE_DAY late. This
        // leads to each window staying open for ONE_DAY after the watermark has passed the end of the
        // window. Any late data will result in an additional pane being fired for that same window.
  
        // The first pane produced will be ON_TIME and the remaining panes will be LATE.
        // To definitely get the last pane when the window closes, use
        // .withAllowedLateness(ONE_DAY, ClosingBehavior.FIRE_ALWAYS).
  
        // The results for the example above with the default trigger and ONE_DAY allowed lateness
        // would be:
        // Key (freeway) | Value (totalFlow) | numberOfRecords | isFirst | isLast | timing
        // 5             | 260                | 6                 | true    | false  | ON_TIME
        // 5             | 60                 | 1                 | false   | false  | LATE
        // 5             | 30                 | 1                 | false   | false  | LATE
        // 5             | 20                 | 1                 | false   | false  | LATE
        // 5             | 60                 | 1                 | false   | false  | LATE
        WindowedStream<SampleBean> withAllowedLatenessResults = inputStream
            .window(new WindowOption.TimeWindows(Duration.standardMinutes(windowDuration)),
            new TriggerOption().discardingFiredPanes(),
            Duration.standardDays(1))
            .addCompositeStreams(new TotalFlow("withAllowedLateness"));
  
        // Concept #3: How to get speculative estimates
        // We can specify a trigger that fires independent of the watermark, for instance after
        // ONE_MINUTE of processing time. This allows us to produce speculative estimates before
        // all the data is available. Since we don't have any triggers that depend on the watermark
        // we don't get an ON_TIME firing. Instead, all panes are either EARLY or LATE.
  
        // We also use accumulatingFiredPanes to build up the results across each pane firing.
  
        // The results for the example above for this trigger would be:
        // Key (freeway) | Value (totalFlow) | numberOfRecords | isFirst | isLast | timing
        // 5             | 80                 | 2                 | true    | false  | EARLY
        // 5             | 100                | 3                 | false   | false  | EARLY
        // 5             | 260                | 6                 | false   | false  | EARLY
        // 5             | 320                | 7                 | false   | false  | LATE
        // 5             | 370                | 9                 | false   | false  | LATE
        // 5             | 430                | 10                | false   | false  | LATE
  
        ApexStream<SampleBean> speculativeResults = inputStream
            .window(new WindowOption.TimeWindows(Duration.standardMinutes(windowDuration)),
                //Trigger fires every minute
            new TriggerOption().withEarlyFiringsAtEvery(Duration.standardMinutes(1))
                    // After emitting each pane, it will continue accumulating the elements so that each
                    // approximation includes all of the previous data in addition to the newly arrived
                    // data.
            .accumulatingFiredPanes(),
            Duration.standardDays(1))
            .addCompositeStreams(new TotalFlow("speculative"));
  
        // Concept #4: Combining late data and speculative estimates
        // We can put the previous concepts together to get EARLY estimates, an ON_TIME result,
        // and LATE updates based on late data.
  
        // Each time a triggering condition is satisfied it advances to the next trigger.
        // If there are new elements this trigger emits a window under following condition:
        // > Early approximations every minute till the end of the window.
        // > An on-time firing when the watermark has passed the end of the window
        // > Every five minutes of late data.
  
        // Every pane produced will either be EARLY, ON_TIME or LATE.
  
        // The results for the example above for this trigger would be:
        // Key (freeway) | Value (totalFlow) | numberOfRecords | isFirst | isLast | timing
        // 5             | 80                 | 2                 | true    | false  | EARLY
        // 5             | 100                | 3                 | false   | false  | EARLY
        // 5             | 260                | 6                 | false   | false  | EARLY
        // [First pane fired after the end of the window]
        // 5             | 320                | 7                 | false   | false  | ON_TIME
        // 5             | 430                | 10                | false   | false  | LATE
  
        // For more possibilities of how to build advanced triggers, see {@link Trigger}.
        WindowedStream<SampleBean> sequentialResults = inputStream
            .window(new WindowOption.TimeWindows(Duration.standardMinutes(windowDuration)),
                // Speculative every ONE_MINUTE
            new TriggerOption().withEarlyFiringsAtEvery(Duration.standardMinutes(1))
            .withLateFiringsAtEvery(Duration.standardMinutes(5))
                    // After emitting each pane, it will continue accumulating the elements so that each
                    // approximation includes all of the previous data in addition to the newly arrived
                    // data.
            .accumulatingFiredPanes(),
            Duration.standardDays(1))
            .addCompositeStreams(new TotalFlow("sequential"));
  
        return sequentialResults;
      }
  
    }
  
    //////////////////////////////////////////////////////////////////////////////////////////////////
    // The remaining parts of the pipeline are needed to produce the output for each
    // concept above. Not directly relevant to understanding the trigger examples.
  
    /**
     * Calculate total flow and number of records for each freeway and format the results to TableRow
     * objects, to save to BigQuery.
     */
    static class TotalFlow extends
        CompositeStreamTransform<WindowedStream<String>, WindowedStream<SampleBean>>
    {
      private String triggerType;
  
      public TotalFlow(String triggerType)
      {
        this.triggerType = triggerType;
      }
  
      @Override
      public WindowedStream<SampleBean> compose(WindowedStream<String> inputStream)
      {
  
        WindowedStream<KeyValPair<String, Iterable<Integer>>> flowPerFreeway = inputStream
            .groupByKey(new ExtractFlowInfo());
  
        return flowPerFreeway
            .map(new Function.MapFunction<KeyValPair<String, Iterable<Integer>>, KeyValPair<String, String>>()
            {
              @Override
              public KeyValPair<String, String> f(KeyValPair<String, Iterable<Integer>> input)
              {
                Iterable<Integer> flows = input.getValue();
                Integer sum = 0;
                Long numberOfRecords = 0L;
                for (Integer value : flows) {
                  sum += value;
                  numberOfRecords++;
                }
                return new KeyValPair<>(input.getKey(), sum + "," + numberOfRecords);
              }
            })
            .map(new FormatTotalFlow(triggerType));
      }
    }
  
    /**
     * Format the results of the Total flow calculation to a TableRow, to save to BigQuery.
     * Adds the triggerType, pane information, processing time and the window timestamp.
     */
    static class FormatTotalFlow implements Function.MapFunction<KeyValPair<String, String>, SampleBean>
    {
      private String triggerType;
  
      public FormatTotalFlow(String triggerType)
      {
        this.triggerType = triggerType;
      }
  
      @Override
      public SampleBean f(KeyValPair<String, String> input)
      {
        String[] values = input.getValue().split(",");
        //TODO need to have a callback to get the metadata like window id, pane id, timestamps etc.
        return new SampleBean(triggerType, input.getKey(), Integer.parseInt(values[0]), Long
            .parseLong(values[1]), null, false, false, null, null, new Date());
      }
    }
  
    public static class SampleBean
    {
      public SampleBean()
      {
      }
  
      private String triggerType;
  
      private String freeway;
  
      private int totalFlow;
  
      private long numberOfRecords;
  
      private String window;
  
      private boolean isFirst;
  
      private boolean isLast;
  
      private Date timing;
  
      private Date eventTime;
  
      private Date processingTime;
  
      @Override
      public boolean equals(Object o)
      {
        if (this == o) {
          return true;
        }
        if (o == null || getClass() != o.getClass()) {
          return false;
        }
        SampleBean that = (SampleBean)o;
        return totalFlow == that.totalFlow &&
            numberOfRecords == that.numberOfRecords &&
            isFirst == that.isFirst &&
            isLast == that.isLast &&
            Objects.equals(triggerType, that.triggerType) &&
            Objects.equals(freeway, that.freeway) &&
            Objects.equals(window, that.window) &&
            Objects.equals(timing, that.timing) &&
            Objects.equals(eventTime, that.eventTime) &&
            Objects.equals(processingTime, that.processingTime);
      }
  
      @Override
      public int hashCode()
      {
        return Objects
            .hash(triggerType, freeway, totalFlow, numberOfRecords, window, isFirst, isLast, timing, eventTime,
              processingTime);
      }
  
      public SampleBean(String triggerType, String freeway, int totalFlow, long numberOfRecords, String window, boolean isFirst, boolean isLast, Date timing, Date eventTime, Date processingTime)
      {
  
        this.triggerType = triggerType;
        this.freeway = freeway;
        this.totalFlow = totalFlow;
        this.numberOfRecords = numberOfRecords;
        this.window = window;
        this.isFirst = isFirst;
        this.isLast = isLast;
        this.timing = timing;
        this.eventTime = eventTime;
        this.processingTime = processingTime;
      }
  
      public String getTriggerType()
      {
        return triggerType;
      }
  
      public void setTriggerType(String triggerType)
      {
        this.triggerType = triggerType;
      }
  
      public String getFreeway()
      {
        return freeway;
      }
  
      public void setFreeway(String freeway)
      {
        this.freeway = freeway;
      }
  
      public int getTotalFlow()
      {
        return totalFlow;
      }
  
      public void setTotalFlow(int totalFlow)
      {
        this.totalFlow = totalFlow;
      }
  
      public long getNumberOfRecords()
      {
        return numberOfRecords;
      }
  
      public void setNumberOfRecords(long numberOfRecords)
      {
        this.numberOfRecords = numberOfRecords;
      }
  
      public String getWindow()
      {
        return window;
      }
  
      public void setWindow(String window)
      {
        this.window = window;
      }
  
      public boolean isFirst()
      {
        return isFirst;
      }
  
      public void setFirst(boolean first)
      {
        isFirst = first;
      }
  
      public boolean isLast()
      {
        return isLast;
      }
  
      public void setLast(boolean last)
      {
        isLast = last;
      }
  
      public Date getTiming()
      {
        return timing;
      }
  
      public void setTiming(Date timing)
      {
        this.timing = timing;
      }
  
      public Date getEventTime()
      {
        return eventTime;
      }
  
      public void setEventTime(Date eventTime)
      {
        this.eventTime = eventTime;
      }
  
      public Date getProcessingTime()
      {
        return processingTime;
      }
  
      public void setProcessingTime(Date processingTime)
      {
        this.processingTime = processingTime;
      }
    }
  
    /**
     * Extract the freeway and total flow in a reading.
     * Freeway is used as key since we are calculating the total flow for each freeway.
     */
    static class ExtractFlowInfo implements Function.ToKeyValue<String, String, Integer>
    {
      @Override
      public Tuple<KeyValPair<String, Integer>> f(String input)
      {
        String[] laneInfo = input.split(",");
        if (laneInfo[0].equals("timestamp")) {
          // Header row
          return null;
        }
        if (laneInfo.length < 48) {
          //Skip the invalid input.
          return null;
        }
        String freeway = laneInfo[2];
        Integer totalFlow = tryIntegerParse(laneInfo[7]);
        // Ignore the records with total flow 0 to easily understand the working of triggers.
        // Skip the records with total flow -1 since they are invalid input.
        if (totalFlow == null || totalFlow <= 0) {
          return null;
        }
        return new Tuple.PlainTuple<>(new KeyValPair<>(freeway, totalFlow));
      }
    }
  
    private static final String PUBSUB_TIMESTAMP_LABEL_KEY = "timestamp_ms";
  
    public static void main(String[] args) throws Exception
    {
      StreamFactory.fromFolder("some folder")
          .addCompositeStreams(new CalculateTotalFlow(60));
  
    }
  
    private static Integer tryIntegerParse(String number)
    {
      try {
        return Integer.parseInt(number);
      } catch (NumberFormatException e) {
        return null;
      }
    }
  
  }
  