datafu-pig/src/main/java/datafu/pig/stats/DoubleVAR.java [107:215]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - t.set(0, null); t.set(1, null); t.set(2, 0L); } else { t.set(0, d); t.set(1, d*d); t.set(2, 1L); } return t; } catch(NumberFormatException nfe) { nfe.printStackTrace(); // invalid input, // treat this input as null try { t.set(0, null); t.set(1, null); t.set(2, 0L); } catch (ExecException e) { throw e; } return t; } catch (ExecException ee) { ee.printStackTrace(); throw ee; } catch (Exception e) { e.printStackTrace(); int errCode = 2106; String msg = "Error while computing variance in " + this.getClass().getSimpleName(); throw new ExecException(msg, errCode, PigException.BUG, e); } } } static public class Intermediate extends EvalFunc { @Override public Tuple exec(Tuple input) throws IOException { try { DataBag b = (DataBag)input.get(0); return combine(b); } catch (ExecException ee) { throw ee; } catch (Exception e) { int errCode = 2106; String msg = "Error while computing variacne in " + this.getClass().getSimpleName(); throw new ExecException(msg, errCode, PigException.BUG, e); } } } static public class Final extends EvalFunc { @Override public Double exec(Tuple input) throws IOException { try { DataBag b = (DataBag)input.get(0); Tuple combined = combine(b); Double sum = (Double)combined.get(0); Double sumSquare = (Double)combined.get(1); if(sum == null) { return null; } Long count = (Long)combined.get(2); Double var = null; if (count > 0) { Double avg = new Double(sum / count); Double avgSquare = new Double(sumSquare / count); var = avgSquare - avg*avg; } return var; } catch (ExecException ee) { throw ee; } catch (Exception e) { int errCode = 2106; String msg = "Error while computing variance in " + this.getClass().getSimpleName(); throw new ExecException(msg, errCode, PigException.BUG, e); } } } static protected Tuple combine(DataBag values) throws ExecException{ double sum = 0; double sumSquare = 0; long totalCount = 0; // combine is called from Intermediate and Final // In either case, Initial would have been called // before and would have sent in valid tuples // Hence we don't need to check if incoming bag // is empty Tuple output = mTupleFactory.newTuple(3); boolean sawNonNull = false; for (Iterator it = values.iterator(); it.hasNext();) { Tuple t = it.next(); Double d = (Double)t.get(0); Double dSquare = (Double)t.get(1); Long count = (Long)t.get(2); // we count nulls in var as contributing 0 // a departure from SQL for performance of // COUNT() which implemented by just inspecting // size of the bag if(d == null) { d = 0.0; - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - datafu-pig/src/main/java/datafu/pig/stats/VAR.java [124:232]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - t.set(0, null); t.set(1, null); t.set(2, 0L); } else { t.set(0, d); t.set(1, d*d); t.set(2, 1L); } return t; } catch(NumberFormatException nfe) { nfe.printStackTrace(); // invalid input, // treat this input as null try { t.set(0, null); t.set(1, null); t.set(2, 0L); } catch (ExecException e) { throw e; } return t; } catch (ExecException ee) { ee.printStackTrace(); throw ee; } catch (Exception e) { e.printStackTrace(); int errCode = 2106; String msg = "Error while computing variance in " + this.getClass().getSimpleName(); throw new ExecException(msg, errCode, PigException.BUG, e); } } } static public class Intermediate extends EvalFunc { @Override public Tuple exec(Tuple input) throws IOException { try { DataBag b = (DataBag)input.get(0); return combine(b); } catch (ExecException ee) { throw ee; } catch (Exception e) { int errCode = 2106; String msg = "Error while computing variacne in " + this.getClass().getSimpleName(); throw new ExecException(msg, errCode, PigException.BUG, e); } } } static public class Final extends EvalFunc { @Override public Double exec(Tuple input) throws IOException { try { DataBag b = (DataBag)input.get(0); Tuple combined = combine(b); Double sum = (Double)combined.get(0); Double sumSquare = (Double)combined.get(1); if(sum == null) { return null; } Long count = (Long)combined.get(2); Double var = null; if (count > 0) { Double avg = new Double(sum / count); Double avgSquare = new Double(sumSquare / count); var = avgSquare - avg*avg; } return var; } catch (ExecException ee) { throw ee; } catch (Exception e) { int errCode = 2106; String msg = "Error while computing variance in " + this.getClass().getSimpleName(); throw new ExecException(msg, errCode, PigException.BUG, e); } } } static protected Tuple combine(DataBag values) throws ExecException{ double sum = 0; double sumSquare = 0; long totalCount = 0; // combine is called from Intermediate and Final // In either case, Initial would have been called // before and would have sent in valid tuples // Hence we don't need to check if incoming bag // is empty Tuple output = mTupleFactory.newTuple(3); boolean sawNonNull = false; for (Iterator it = values.iterator(); it.hasNext();) { Tuple t = it.next(); Double d = (Double)t.get(0); Double dSquare = (Double)t.get(1); Long count = (Long)t.get(2); // we count nulls in var as contributing 0 // a departure from SQL for performance of // COUNT() which implemented by just inspecting // size of the bag if(d == null) { d = 0.0; - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -