java/pig-galago/src/com/yahoo/pig/builtin/SUM.java

   1 /*
   2  * Copyright (c) 2007 Yahoo! Inc.  All rights reserved.
   3  * See accompanying LICENSE file.
   4  */
   5 package com.yahoo.pig.builtin;
   6
   7 import java.io.IOException;
   8 import java.util.Iterator;
   9
  10 import com.yahoo.pig.AlgebraicAtomEvalFunc;
  11 import com.yahoo.pig.AtomEvalFunc;
  12 import com.yahoo.pig.TupleEvalFunc;
  13 import com.yahoo.pig.data.DataAtom;
  14 import com.yahoo.pig.data.DataBag;
  15 import com.yahoo.pig.data.Tuple;
  16 import com.yahoo.pig.impl.logicalLayer.schema.SchemaField;
  17 import com.yahoo.pig.impl.logicalLayer.schema.SchemaItem;
  18 import com.yahoo.pig.impl.logicalLayer.schema.SchemaItemList;
  19
  20 /**
  21  * Generates the sum of the values of the first field of a tuple.
  22  */
  23 public class SUM extends AlgebraicAtomEvalFunc {
  24
  25     @Override
  26     public void exec(Tuple input, DataAtom output) throws IOException {
  27         output.setValue(sum(input));
  28     }
  29
  30     public TupleEvalFunc getInitial() {
  31         return new Initial();
  32     }
  33
  34     public TupleEvalFunc getIntermed() {
  35         return new Initial();
  36     }
  37
  38     public AtomEvalFunc getFinal() {
  39         return new Final();
  40     }
  41
  42     static public class Initial extends TupleEvalFunc {
  43         public void exec(Tuple input, Tuple output) throws IOException {
  44             output.appendField(new DataAtom(sum(input)));
  45         }
  46     }
  47     static public class Final extends AtomEvalFunc {
  48         public void exec(Tuple input, DataAtom output) throws IOException {
  49             output.setValue(sum(input));
  50         }
  51     }
  52
  53     static protected double sum(Tuple input) throws IOException {
  54         DataBag values = input.getBagField(0);
  55
  56         double sum = 0;
  57         for (Iterator it = values.content(); it.hasNext();) {
  58             Tuple t = (Tuple) it.next();
  59             try {
  60             sum += t.getAtomField(0).numval();
  61             }catch(RuntimeException exp) {
  62                 throw new RuntimeException(exp.getMessage() + " error processing: " + t.toString());
  63             }
  64         }
  65
  66         return sum;
  67     }
  68     @Override
  69     public SchemaItem outputSchema() {
  70         return new SchemaField("sum");
  71     }
  72 }