java/pig-galago/src/com/yahoo/pig/impl/eval/window/TimeWindowSpec.java

   1 /*
   2  * Copyright (c) 2007 Yahoo! Inc.  All rights reserved.
   3  * See accompanying LICENSE file.
   4  */
   5 package com.yahoo.pig.impl.eval.window;
   6
   7 import java.io.IOException;
   8 import java.util.Iterator;
   9 import java.util.LinkedList;
  10 import java.util.List;
  11
  12 import com.yahoo.pig.data.DataCollector;
  13 import com.yahoo.pig.data.TimestampedTuple;
  14 import com.yahoo.pig.data.Tuple;
  15
  16 public class TimeWindowSpec extends WindowSpec {
  17         double duration;  // duration in seconds
  18         List<TimestampedTuple> window;
  19
  20         public TimeWindowSpec(windowType type, double duration){
  21                 super(type);
  22                 this.duration = duration;
  23         window = new LinkedList<TimestampedTuple>();
  24     }
  25
  26     @Override
  27     public DataCollector collector(final DataCollector output) {
  28         return new DataCollector() {
  29
  30             public void add(Tuple t) throws IOException {
  31
  32                 boolean changed = false;
  33
  34                 if (!(((TimestampedTuple) t).isHeartbeat)) {
  35                     window.add((TimestampedTuple) t);
  36                     changed = true;
  37                 }
  38
  39                 double expireTime = ((TimestampedTuple) t).timestamp - duration;
  40                 while (true) {
  41                     TimestampedTuple tail = window.get(0);
  42
  43                     if (tail != null & tail.timestamp <= expireTime) {
  44                         window.remove(0);
  45                         changed = true;
  46                     } else {
  47                         break;
  48                     }
  49                 }
  50
  51                 if (changed) {
  52                     // emit entire window content to output collector
  53                     for (Iterator<TimestampedTuple> it = window.iterator(); it.hasNext(); ) {
  54                         output.add(it.next());
  55                     }
  56                 }
  57             }
  58         };
  59     }
  60 }