apex-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From t..@apache.org
Subject [08/30] apex-malhar git commit: Renamed demos to examples. Packages and artifactid names are changed as suggested.
Date Tue, 07 Mar 2017 06:58:13 GMT
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/Application.java
----------------------------------------------------------------------
diff --git a/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/Application.java b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/Application.java
new file mode 100644
index 0000000..8132c7d
--- /dev/null
+++ b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/Application.java
@@ -0,0 +1,83 @@
+/**
+ * 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.examples.machinedata;
+
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.apex.examples.machinedata.data.MachineKey;
+import org.apache.apex.examples.machinedata.operator.MachineInfoAveragingOperator;
+import org.apache.apex.examples.machinedata.operator.MachineInfoAveragingPrerequisitesOperator;
+import org.apache.hadoop.conf.Configuration;
+
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.api.annotation.ApplicationAnnotation;
+import com.datatorrent.contrib.redis.RedisKeyValPairOutputOperator;
+import com.datatorrent.lib.io.SmtpOutputOperator;
+
+/**
+ * <p>
+ * Resource monitor application.
+ * </p>
+ *
+ * @since 0.3.5
+ */
+@ApplicationAnnotation(name = "MachineDataExample")
+@SuppressWarnings("unused")
+public class Application implements StreamingApplication
+{
+
+  private static final Logger LOG = LoggerFactory.getLogger(Application.class);
+
+  /**
+   * This function sets up the DAG for calculating the average
+   *
+   * @param dag  the DAG instance
+   * @param conf the configuration instance
+   * @return MachineInfoAveragingPrerequisitesOperator
+   */
+  private MachineInfoAveragingPrerequisitesOperator addAverageCalculation(DAG dag, Configuration conf)
+  {
+    MachineInfoAveragingPrerequisitesOperator prereqAverageOper = dag.addOperator("Aggregator", MachineInfoAveragingPrerequisitesOperator.class);
+    MachineInfoAveragingOperator averageOperator = dag.addOperator("AverageCalculator", MachineInfoAveragingOperator.class);
+    RedisKeyValPairOutputOperator<MachineKey, Map<String, String>> redisAvgOperator = dag.addOperator("Persister", new RedisKeyValPairOutputOperator<MachineKey, Map<String, String>>());
+    dag.addStream("Average", averageOperator.outputPort, redisAvgOperator.input);
+    SmtpOutputOperator smtpOutputOperator = dag.addOperator("Alerter", new SmtpOutputOperator());
+    dag.addStream("Aggregates", prereqAverageOper.outputPort, averageOperator.inputPort);
+    dag.addStream("Alerts", averageOperator.smtpAlert, smtpOutputOperator.input);
+    return prereqAverageOper;
+  }
+
+  /**
+   * Create the DAG
+   */
+  @Override
+  public void populateDAG(DAG dag, Configuration conf)
+  {
+    InputReceiver randomGen = dag.addOperator("Receiver", InputReceiver.class);
+    DimensionGenerator dimensionGenerator = dag.addOperator("DimensionsGenerator", DimensionGenerator.class);
+    dag.addStream("Events", randomGen.outputInline, dimensionGenerator.inputPort);
+    MachineInfoAveragingPrerequisitesOperator prereqAverageOper = addAverageCalculation(dag, conf);
+    dag.addStream("DimensionalData", dimensionGenerator.outputInline, prereqAverageOper.inputPort);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/DimensionGenerator.java
----------------------------------------------------------------------
diff --git a/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/DimensionGenerator.java b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/DimensionGenerator.java
new file mode 100644
index 0000000..6c717c2
--- /dev/null
+++ b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/DimensionGenerator.java
@@ -0,0 +1,121 @@
+/**
+ * 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.examples.machinedata;
+
+import org.apache.apex.examples.machinedata.data.MachineInfo;
+import org.apache.apex.examples.machinedata.data.MachineKey;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.common.util.BaseOperator;
+
+
+/**
+ * <p>
+ * Information tuple generator with randomness.
+ * </p>
+ *
+ * @since 0.3.5
+ */
+@SuppressWarnings("unused")
+public class DimensionGenerator extends BaseOperator
+{
+  public transient DefaultOutputPort<MachineInfo> outputInline = new DefaultOutputPort<>();
+  public transient DefaultOutputPort<MachineInfo> output = new DefaultOutputPort<>();
+  private int threshold = 90;
+
+  public final transient DefaultInputPort<MachineInfo> inputPort = new DefaultInputPort<MachineInfo>()
+  {
+
+    @Override
+    public void process(MachineInfo tuple)
+    {
+      emitDimensions(tuple);
+    }
+
+  };
+
+  @Override
+  public void setup(Context.OperatorContext context)
+  {
+    super.setup(context);
+  }
+
+  /**
+   * This returns the threshold value set
+   * @return
+   */
+  public int getThreshold()
+  {
+    return threshold;
+  }
+
+  /**
+   * This function sets the threshold value. This value is used to check the maximum value for cpu/ram/hdd
+   * @param threshold
+   */
+  public void setThreshold(int threshold)
+  {
+    this.threshold = threshold;
+  }
+
+  /**
+   * This function takes in the tuple from upstream operator and generates tuples with different dimension combinations
+   *
+   * @param tuple
+   */
+  private void emitDimensions(MachineInfo tuple)
+  {
+    MachineKey tupleKey = tuple.getMachineKey();
+
+    for (int i = 0; i < 64; i++) {
+      MachineKey machineKey = new MachineKey(tupleKey.getTimeKey(),tupleKey.getDay());
+      if ((i & 1) != 0) {
+        machineKey.setCustomer(tupleKey.getCustomer());
+      }
+      if ((i & 2) != 0) {
+        machineKey.setProduct(tupleKey.getProduct());
+      }
+      if ((i & 4) != 0) {
+        machineKey.setOs(tupleKey.getOs());
+      }
+      if ((i & 8) != 0) {
+        machineKey.setDeviceId(tupleKey.getDeviceId());
+      }
+      if ((i & 16) != 0) {
+        machineKey.setSoftware1(tupleKey.getSoftware1());
+      }
+      if ((i & 32) != 0) {
+        machineKey.setSoftware2(tupleKey.getSoftware2());
+      }
+
+      int cpu = tuple.getCpu();
+      int ram = tuple.getRam();
+      int hdd = tuple.getHdd();
+      MachineInfo machineInfo = new MachineInfo();
+      machineInfo.setMachineKey(machineKey);
+      machineInfo.setCpu((cpu < threshold) ? cpu : threshold);
+      machineInfo.setRam((ram < threshold) ? ram : threshold);
+      machineInfo.setHdd((hdd < threshold) ? hdd : threshold);
+      outputInline.emit(machineInfo);
+      output.emit(machineInfo);
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/InputReceiver.java
----------------------------------------------------------------------
diff --git a/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/InputReceiver.java b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/InputReceiver.java
new file mode 100644
index 0000000..263db55
--- /dev/null
+++ b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/InputReceiver.java
@@ -0,0 +1,524 @@
+/**
+ * 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.examples.machinedata;
+
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.Random;
+import java.util.TimeZone;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.apex.examples.machinedata.data.MachineInfo;
+import org.apache.apex.examples.machinedata.data.MachineKey;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.InputOperator;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * <p>
+ * Information tuple generator with randomness.
+ * </p>
+ *
+ * @since 0.3.5
+ */
+@SuppressWarnings("unused")
+public class InputReceiver extends BaseOperator implements InputOperator
+{
+  private static final Logger logger = LoggerFactory.getLogger(InputReceiver.class);
+
+  public transient DefaultOutputPort<MachineInfo> outputInline = new DefaultOutputPort<>();
+  private final Random randomGen = new Random();
+
+  private int customerMin = 1;
+  private int customerMax = 5;
+  private int productMin = 4;
+  private int productMax = 6;
+  private int osMin = 10;
+  private int osMax = 12;
+  private int software1Min = 10;
+  private int software1Max = 12;
+  private int software2Min = 12;
+  private int software2Max = 14;
+  private int software3Min = 4;
+  private int software3Max = 6;
+
+  private int deviceIdMin = 1;
+  private int deviceIdMax = 50;
+
+  private int tupleBlastSize = 1001;
+
+  private static final DateFormat minuteDateFormat = new SimpleDateFormat("HHmm");
+  private static final DateFormat dayDateFormat = new SimpleDateFormat("d");
+
+  static {
+    TimeZone tz = TimeZone.getTimeZone("GMT");
+    minuteDateFormat.setTimeZone(tz);
+    dayDateFormat.setTimeZone(tz);
+
+  }
+
+  @Override
+  public void setup(Context.OperatorContext context)
+  {
+    super.setup(context);
+  }
+
+  @Override
+  public void beginWindow(long windowId)
+  {
+    super.beginWindow(windowId);
+  }
+
+  @Override
+  public void emitTuples()
+  {
+    int count = 0;
+    Calendar calendar = Calendar.getInstance();
+    Date date = calendar.getTime();
+    String timeKey = minuteDateFormat.format(date);
+    String day = dayDateFormat.format(date);
+
+    while (count < tupleBlastSize) {
+      randomGen.setSeed(System.currentTimeMillis());
+
+      int customerVal = genCustomerId();
+      int productVal = genProductVer();
+      int osVal = genOsVer();
+      int software1Val = genSoftware1Ver();
+      int software2Val = genSoftware2Ver();
+      int software3Val = genSoftware3Ver();
+      int deviceIdVal = genDeviceId();
+
+      int cpuVal = genCpu(calendar);
+      int ramVal = genRam(calendar);
+      int hddVal = genHdd(calendar);
+
+      MachineKey machineKey = new MachineKey(timeKey, day);
+
+      machineKey.setCustomer(customerVal);
+      machineKey.setProduct(productVal);
+      machineKey.setOs(osVal);
+      machineKey.setDeviceId(deviceIdVal);
+      machineKey.setSoftware1(software1Val);
+      machineKey.setSoftware2(software2Val);
+      machineKey.setSoftware3(software3Val);
+      MachineInfo machineInfo = new MachineInfo();
+      machineInfo.setMachineKey(machineKey);
+      machineInfo.setCpu(cpuVal);
+      machineInfo.setRam(ramVal);
+      machineInfo.setHdd(hddVal);
+
+      outputInline.emit(machineInfo);
+
+      count++;
+    }
+  }
+
+  private int genCustomerId()
+  {
+    int range = customerMax - customerMin + 1;
+    return customerMin + randomGen.nextInt(range);
+  }
+
+  private int genProductVer()
+  {
+    int range = productMax - productMin + 1;
+    return productMin + randomGen.nextInt(range);
+  }
+
+  private int genOsVer()
+  {
+    int range = osMax - osMin + 1;
+    return osMin + randomGen.nextInt(range);
+  }
+
+  private int genSoftware3Ver()
+  {
+    int range = software3Max - software3Min + 1;
+    return software3Min + randomGen.nextInt(range);
+  }
+
+  private int genDeviceId()
+  {
+    int range = deviceIdMax - deviceIdMin + 1;
+    return deviceIdMin + randomGen.nextInt(range);
+  }
+
+  private int genSoftware1Ver()
+  {
+    int range = software1Max - software1Min + 1;
+    return software1Min + randomGen.nextInt(range);
+  }
+
+  private int genSoftware2Ver()
+  {
+    int range = software2Max - software2Min + 1;
+    return software2Min + randomGen.nextInt(range);
+  }
+
+  private int genCpu(Calendar cal)
+  {
+    int minute = cal.get(Calendar.MINUTE);
+    int second;
+    int range = minute / 2 + 19;
+    if (minute / 17 == 0) {
+      second = cal.get(Calendar.SECOND);
+      return (30 + randomGen.nextInt(range) + (minute % 7) - (second % 11));
+    } else if (minute / 47 == 0) {
+      second = cal.get(Calendar.SECOND);
+      return (7 + randomGen.nextInt(range) + (minute % 7) - (second % 7));
+    } else {
+      second = cal.get(Calendar.SECOND);
+      return (randomGen.nextInt(range) + (minute % 19) + (second % 7));
+    }
+  }
+
+  private int genRam(Calendar cal)
+  {
+    int minute = cal.get(Calendar.MINUTE);
+    int second;
+    int range = minute + 1;
+    if (minute / 23 == 0) {
+      second = cal.get(Calendar.SECOND);
+      return (20 + randomGen.nextInt(range) + (minute % 5) - (second % 11));
+    } else if (minute / 37 == 0) {
+      second = cal.get(Calendar.SECOND);
+      return (11 + randomGen.nextInt(60) - (minute % 5) - (second % 11));
+    } else {
+      second = cal.get(Calendar.SECOND);
+      return (randomGen.nextInt(range) + (minute % 17) + (second % 11));
+    }
+  }
+
+  private int genHdd(Calendar cal)
+  {
+    int minute = cal.get(Calendar.MINUTE);
+    int second;
+    int range = minute / 2 + 1;
+    if (minute / 37 == 0) {
+      second = cal.get(Calendar.SECOND);
+      return (25 + randomGen.nextInt(range) - minute % 7 - second % 11);
+    } else {
+      second = cal.get(Calendar.SECOND);
+      return (randomGen.nextInt(range) + minute % 23 + second % 11);
+    }
+  }
+
+  /**
+   * This method returns the minimum value for customer
+   *
+   * @return
+   */
+  public int getCustomerMin()
+  {
+    return customerMin;
+  }
+
+  /**
+   * This method is used to set the minimum value for customer
+   *
+   * @param customerMin the minimum customer value
+   */
+  public void setCustomerMin(int customerMin)
+  {
+    this.customerMin = customerMin;
+  }
+
+  /**
+   * This method returns the max value for customer
+   *
+   * @return
+   */
+  public int getCustomerMax()
+  {
+    return customerMax;
+  }
+
+  /**
+   * This method is used to set the max value for customer
+   *
+   * @param customerMax the max customer value
+   */
+  public void setCustomerMax(int customerMax)
+  {
+    this.customerMax = customerMax;
+  }
+
+  /**
+   * This method returns the minimum value for product
+   *
+   * @return
+   */
+  public int getProductMin()
+  {
+    return productMin;
+  }
+
+  /**
+   * This method is used to set the minimum value for product
+   *
+   * @param productMin the minimum product value
+   */
+  public void setProductMin(int productMin)
+  {
+    this.productMin = productMin;
+  }
+
+  /**
+   * This method returns the max value for product
+   *
+   * @return
+   */
+  public int getProductMax()
+  {
+    return productMax;
+  }
+
+  /**
+   * This method is used to set the max value for product
+   *
+   * @param productMax the max product value
+   */
+  public void setProductMax(int productMax)
+  {
+    this.productMax = productMax;
+  }
+
+  /**
+   * This method returns the minimum value for OS
+   *
+   * @return
+   */
+  public int getOsMin()
+  {
+    return osMin;
+  }
+
+  /**
+   * This method is used to set the minimum value for OS
+   *
+   * @param osMin the min OS value
+   */
+  public void setOsMin(int osMin)
+  {
+    this.osMin = osMin;
+  }
+
+  /**
+   * This method returns the max value for OS
+   *
+   * @return
+   */
+  public int getOsMax()
+  {
+    return osMax;
+  }
+
+  /**
+   * This method is used to set the max value for OS
+   *
+   * @param osMax the max OS value
+   */
+  public void setOsMax(int osMax)
+  {
+    this.osMax = osMax;
+  }
+
+  /**
+   * This method returns the minimum value for software1
+   *
+   * @return
+   */
+  public int getSoftware1Min()
+  {
+    return software1Min;
+  }
+
+  /**
+   * This method is used to set the minimum value for software1
+   *
+   * @param software1Min the minimum software1 value
+   */
+  public void setSoftware1Min(int software1Min)
+  {
+    this.software1Min = software1Min;
+  }
+
+  /**
+   * This method returns the max value for software1
+   *
+   * @return
+   */
+  public int getSoftware1Max()
+  {
+    return software1Max;
+  }
+
+  /**
+   * This method is used to set the max value for software1
+   *
+   * @param software1Max the max software1 value
+   */
+  public void setSoftware1Max(int software1Max)
+  {
+    this.software1Max = software1Max;
+  }
+
+  /**
+   * This method returns the minimum value for software2
+   *
+   * @return
+   */
+  public int getSoftware2Min()
+  {
+    return software2Min;
+  }
+
+  /**
+   * This method is used to set the minimum value for software2
+   *
+   * @param software2Min the minimum software2 value
+   */
+  public void setSoftware2Min(int software2Min)
+  {
+    this.software2Min = software2Min;
+  }
+
+  /**
+   * This method returns the max value for software2
+   *
+   * @return
+   */
+  public int getSoftware2Max()
+  {
+    return software2Max;
+  }
+
+  /**
+   * This method is used to set the max value for software2
+   *
+   * @param software2Max the max software2 value
+   */
+  public void setSoftware2Max(int software2Max)
+  {
+    this.software2Max = software2Max;
+  }
+
+  /**
+   * This method returns the minimum value for software3
+   *
+   * @return
+   */
+  public int getSoftware3Min()
+  {
+    return software3Min;
+  }
+
+  /**
+   * This method is used to set the minimum value for software3
+   *
+   * @param software3Min the minimum software3 value
+   */
+  public void setSoftware3Min(int software3Min)
+  {
+    this.software3Min = software3Min;
+  }
+
+  /**
+   * This method returns the max value for software3
+   *
+   * @return
+   */
+  public int getSoftware3Max()
+  {
+    return software3Max;
+  }
+
+  /**
+   * This method is used to set the max value for software3
+   *
+   * @param software3Max the max software3 value
+   */
+  public void setSoftware3Max(int software3Max)
+  {
+    this.software3Max = software3Max;
+  }
+
+  /**
+   * This method returns the minimum value for deviceId
+   *
+   * @return
+   */
+  public int getDeviceIdMin()
+  {
+    return deviceIdMin;
+  }
+
+  /**
+   * This method is used to set the minimum value for deviceId
+   *
+   * @param deviceIdMin the minimum deviceId value
+   */
+  public void setDeviceIdMin(int deviceIdMin)
+  {
+    this.deviceIdMin = deviceIdMin;
+  }
+
+  /**
+   * This method returns the max value for deviceId
+   *
+   * @return
+   */
+  public int getDeviceIdMax()
+  {
+    return deviceIdMax;
+  }
+
+  /**
+   * This method is used to set the max value for deviceId
+   *
+   * @param deviceIdMax the max deviceId value
+   */
+  public void setDeviceIdMax(int deviceIdMax)
+  {
+    this.deviceIdMax = deviceIdMax;
+  }
+
+  /**
+   * @return the tupleBlastSize
+   */
+  public int getTupleBlastSize()
+  {
+    return tupleBlastSize;
+  }
+
+  /**
+   * @param tupleBlastSize the tupleBlastSize to set
+   */
+  public void setTupleBlastSize(int tupleBlastSize)
+  {
+    this.tupleBlastSize = tupleBlastSize;
+  }
+}

http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/data/AverageData.java
----------------------------------------------------------------------
diff --git a/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/data/AverageData.java b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/data/AverageData.java
new file mode 100644
index 0000000..3c12335
--- /dev/null
+++ b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/data/AverageData.java
@@ -0,0 +1,107 @@
+/**
+ * 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.examples.machinedata.data;
+
+
+/**
+ * This class stores the value of sum and the count of values summed.
+ * <p>
+ * AverageData class.
+ * </p>
+ *
+ * @since 0.3.5
+ */
+public class AverageData
+{
+
+  private long cpu;
+  private long hdd;
+  private long ram;
+  private long count;
+
+  /**
+   * This is default constructor that sets the sum and count to 0
+   */
+  public AverageData()
+  {
+
+  }
+
+  /**
+   * This constructor takes the value of sum and count and initialize the local attributes to corresponding values
+   *
+   * @param count
+   *          the value of count
+   */
+  public AverageData(long cpu,long hdd,long ram, long count)
+  {
+    this.cpu = cpu;
+    this.ram = ram;
+    this.hdd = hdd;
+    this.count = count;
+  }
+
+  public long getCpu()
+  {
+    return cpu;
+  }
+
+  public void setCpu(long cpu)
+  {
+    this.cpu = cpu;
+  }
+
+  public long getHdd()
+  {
+    return hdd;
+  }
+
+  public void setHdd(long hdd)
+  {
+    this.hdd = hdd;
+  }
+
+  public long getRam()
+  {
+    return ram;
+  }
+
+  public void setRam(long ram)
+  {
+    this.ram = ram;
+  }
+
+  /**
+   * This returns the value of count
+   * @return
+   */
+  public long getCount()
+  {
+    return count;
+  }
+
+  /**
+   * This method sets the value of count
+   * @param count
+   */
+  public void setCount(long count)
+  {
+    this.count = count;
+  }
+}

http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/data/MachineInfo.java
----------------------------------------------------------------------
diff --git a/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/data/MachineInfo.java b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/data/MachineInfo.java
new file mode 100644
index 0000000..3952b70
--- /dev/null
+++ b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/data/MachineInfo.java
@@ -0,0 +1,158 @@
+/**
+ * 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.examples.machinedata.data;
+
+/**
+ * This class stores the cpu% usage, ram% usage, hdd% usage and key information about a particular machine
+ * <p>
+ * MachineInfo class.
+ * </p>
+ *
+ * @since 0.3.5
+ */
+public class MachineInfo
+{
+  private MachineKey machineKey;
+  private int cpu;
+  private int ram;
+  private int hdd;
+
+  /**
+   * This default constructor
+   */
+  public MachineInfo()
+  {
+  }
+
+  /**
+   * This constructor takes MachineKey as input and initialize local attributes
+   *
+   * @param machineKey
+   *          the MachineKey instance
+   */
+  public MachineInfo(MachineKey machineKey)
+  {
+    this.machineKey = machineKey;
+  }
+
+  /**
+   * This constructor takes MachineKey, cpu usage, ram usage, hdd usage as input and initialize local attributes
+   *
+   * @param machineKey
+   *          the MachineKey instance
+   * @param cpu
+   *          the CPU% usage
+   * @param ram
+   *          the RAM% usage
+   * @param hdd
+   *          the HDD% usage
+   */
+  public MachineInfo(MachineKey machineKey, int cpu, int ram, int hdd)
+  {
+    this.machineKey = machineKey;
+    this.cpu = cpu;
+    this.ram = ram;
+    this.hdd = hdd;
+  }
+
+  /**
+   * This method returns the MachineKey
+   *
+   * @return
+   */
+  public MachineKey getMachineKey()
+  {
+    return machineKey;
+  }
+
+  /**
+   * This method sets the MachineKey
+   *
+   * @param machineKey
+   *          the MachineKey instance
+   */
+  public void setMachineKey(MachineKey machineKey)
+  {
+    this.machineKey = machineKey;
+  }
+
+  /**
+   * This method returns the CPU% usage
+   *
+   * @return
+   */
+  public int getCpu()
+  {
+    return cpu;
+  }
+
+  /**
+   * This method sets the CPU% usage
+   *
+   * @param cpu
+   *          the CPU% usage
+   */
+  public void setCpu(int cpu)
+  {
+    this.cpu = cpu;
+  }
+
+  /**
+   * This method returns the RAM% usage
+   *
+   * @return
+   */
+  public int getRam()
+  {
+    return ram;
+  }
+
+  /**
+   * This method sets the RAM% usage
+   *
+   * @param ram
+   *          the RAM% usage
+   */
+  public void setRam(int ram)
+  {
+    this.ram = ram;
+  }
+
+  /**
+   * This method returns the HDD% usage
+   *
+   * @return
+   */
+  public int getHdd()
+  {
+    return hdd;
+  }
+
+  /**
+   * This method sets the HDD% usage
+   *
+   * @param hdd
+   *          the HDD% usage
+   */
+  public void setHdd(int hdd)
+  {
+    this.hdd = hdd;
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/data/MachineKey.java
----------------------------------------------------------------------
diff --git a/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/data/MachineKey.java b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/data/MachineKey.java
new file mode 100644
index 0000000..2bf0a53
--- /dev/null
+++ b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/data/MachineKey.java
@@ -0,0 +1,381 @@
+/**
+ * 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.examples.machinedata.data;
+
+/**
+ * This class stores the information about various softwares, deviceIds, OS of the device
+ * <p>
+ * MachineKey class.
+ * </p>
+ *
+ * @since 0.3.5
+ */
+public class MachineKey
+{
+
+  private Integer customer;
+  private Integer product;
+  private Integer os;
+  private Integer software1;
+  private Integer software2;
+  private Integer software3;
+  private Integer deviceId;
+  private String timeKey;
+  private String day;
+
+  /**
+   * This constructor takes the format in which time has to be captured and the day when this instance is created
+   *
+   * @param timeKey the format in which time has to be captured
+   * @param day the day when this instance is created
+   */
+  public MachineKey(String timeKey, String day)
+  {
+    this.timeKey = timeKey;
+    this.day = day;
+  }
+
+  /**
+   * This is default constructor
+   */
+  public MachineKey()
+  {
+  }
+
+  /**
+   * This constructor takes format in which time has to be captured, the day when this instance is created, the customer
+   * id, product Id on the device, OS version on the device, software1 version on the device, software2 version on the device,
+   * software3 version on the device, deviceId on the device,
+   *
+   * @param timeKey the format in which time has to be captured
+   * @param day the day when this instance is created
+   * @param customer the customer Id
+   * @param product product Id
+   * @param os OS version
+   * @param software1 software1 version
+   * @param software2 software2 version
+   * @param software3 software3 version
+   * @param deviceId deviceId
+   */
+  public MachineKey(String timeKey, String day, Integer customer, Integer product, Integer os, Integer software1, Integer software2, Integer software3, Integer deviceId)
+  {
+    this.timeKey = timeKey;
+    this.day = day;
+    this.customer = customer;
+    this.product = product;
+    this.os = os;
+    this.software1 = software1;
+    this.software2 = software2;
+    this.software3 = software3;
+    this.deviceId = deviceId;
+  }
+
+  /**
+   * This method returns the format in which the time is captured. The time is the time when this instance of MachineKey
+   * was generated. For e.g. HHmm to capture Hour and minute
+   *
+   * @return
+   */
+  public String getTimeKey()
+  {
+    return timeKey;
+  }
+
+  /**
+   * This method sets the format in which the time is captured. The time is the time when this instance of MachineKey
+   * was generated. For e.g. HHmm to capture Hour and minute
+   *
+   * @param timeKey
+   *          the value of format
+   */
+  public void setTimeKey(String timeKey)
+  {
+    this.timeKey = timeKey;
+  }
+
+  /**
+   * This method returns the day of the month when this instance of MachineKey was generated
+   *
+   * @return
+   */
+  public String getDay()
+  {
+    return day;
+  }
+
+  /**
+   * This method sets the day of the month when this instance of MachineKey was generated
+   *
+   * @param day
+   *          the day of the month
+   */
+  public void setDay(String day)
+  {
+    this.day = day;
+  }
+
+  /**
+   * This method returns the customer Id
+   *
+   * @return
+   */
+  public Integer getCustomer()
+  {
+    return customer;
+  }
+
+  /**
+   * This method sets the customer Id
+   *
+   * @param customer
+   *          the customer Id
+   */
+  public void setCustomer(Integer customer)
+  {
+    this.customer = customer;
+  }
+
+  /**
+   * This method returns product on the device
+   *
+   * @return
+   */
+  public Integer getProduct()
+  {
+    return product;
+  }
+
+  /**
+   * This method sets the product on the device
+   *
+   * @param product
+   *          the value of product
+   */
+  public void setProduct(Integer product)
+  {
+    this.product = product;
+  }
+
+  /**
+   * This method returns the OS version on the device
+   *
+   * @return
+   */
+  public Integer getOs()
+  {
+    return os;
+  }
+
+  /**
+   * This method sets the OS version on the device
+   *
+   * @param os
+   *          OS version
+   */
+  public void setOs(Integer os)
+  {
+    this.os = os;
+  }
+
+  /**
+   * This method returns the version of the software1 on the device
+   *
+   * @return
+   */
+  public Integer getSoftware1()
+  {
+    return software1;
+  }
+
+  /**
+   * This method sets the version of the software1 on the device
+   *
+   * @param software1 the version of the software1
+   */
+  public void setSoftware1(Integer software1)
+  {
+    this.software1 = software1;
+  }
+
+  /**
+   * This method returns the version of the software2 on the device
+   *
+   * @return
+   */
+  public Integer getSoftware2()
+  {
+    return software2;
+  }
+
+  /**
+   * This method sets the version of the software2 on the device
+   *
+   * @param software2
+   *          the version of the software2
+   */
+  public void setSoftware2(Integer software2)
+  {
+    this.software2 = software2;
+  }
+
+  /**
+   * This method returns the version of the software3 on the device
+   *
+   * @return
+   */
+  public Integer getSoftware3()
+  {
+    return software3;
+  }
+
+  /**
+   * This method sets the version of the software3 on the device
+   *
+   * @param software3
+   *          the version of the software3
+   */
+  public void setSoftware3(Integer software3)
+  {
+    this.software3 = software3;
+  }
+
+  @Override
+  public int hashCode()
+  {
+    int key = 0;
+    if (customer != null) {
+      key |= (1 << 31);
+      key ^= customer;
+    }
+    if (product != null) {
+      key |= (1 << 30);
+      key ^= product;
+    }
+    if (os != null) {
+      key |= (1 << 29);
+      key ^= os;
+    }
+    if (software1 != null) {
+      key |= (1 << 28);
+      key ^= software1;
+    }
+    if (software2 != null) {
+      key |= (1 << 27);
+      key ^= software2;
+    }
+    if (software3 != null) {
+      key |= (1 << 26);
+      key ^= software3;
+    }
+    if (deviceId != null) {
+      key |= (1 << 25);
+      key ^= deviceId;
+    }
+    if (timeKey != null) {
+      key |= (1 << 24);
+      key ^= timeKey.hashCode();
+    }
+    if (day != null) {
+      key |= (1 << 23);
+      key ^= day.hashCode();
+    }
+
+    return key;
+  }
+
+  @Override
+  public boolean equals(Object obj)
+  {
+    if (!(obj instanceof MachineKey)) {
+      return false;
+    }
+    MachineKey mkey = (MachineKey)obj;
+    return checkStringEqual(this.timeKey, mkey.timeKey) && checkStringEqual(this.day, mkey.day) && checkIntEqual(this.customer, mkey.customer) && checkIntEqual(this.product, mkey.product) && checkIntEqual(this.os, mkey.os) && checkIntEqual(this.software1, mkey.software1) && checkIntEqual(this.software2, mkey.software2) && checkIntEqual(this.software3, mkey.software3) && checkIntEqual(this.deviceId, mkey.deviceId);
+  }
+
+  private boolean checkIntEqual(Integer a, Integer b)
+  {
+    if ((a == null) && (b == null)) {
+      return true;
+    }
+    if ((a != null) && a.equals(b)) {
+      return true;
+    }
+    return false;
+  }
+
+  private boolean checkStringEqual(String a, String b)
+  {
+    if ((a == null) && (b == null)) {
+      return true;
+    }
+    if ((a != null) && a.equals(b)) {
+      return true;
+    }
+    return false;
+  }
+
+  @Override
+  public String toString()
+  {
+    StringBuilder sb = new StringBuilder(timeKey);
+    if (customer != null) {
+      sb.append("|0:").append(customer);
+    }
+    if (product != null) {
+      sb.append("|1:").append(product);
+    }
+    if (os != null) {
+      sb.append("|2:").append(os);
+    }
+    if (software1 != null) {
+      sb.append("|3:").append(software1);
+    }
+    if (software2 != null) {
+      sb.append("|4:").append(software2);
+    }
+    if (software3 != null) {
+      sb.append("|5:").append(software3);
+    }
+    if (deviceId != null) {
+      sb.append("|6:").append(deviceId);
+    }
+    return sb.toString();
+  }
+
+  /**
+   * This method returns the deviceId of the device
+   * @return The deviceId
+   */
+  public Integer getDeviceId()
+  {
+    return deviceId;
+  }
+
+  /**
+   * This method sets the deviceId of the device
+   *
+   * @param deviceId
+   */
+  public void setDeviceId(Integer deviceId)
+  {
+    this.deviceId = deviceId;
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/data/ResourceType.java
----------------------------------------------------------------------
diff --git a/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/data/ResourceType.java b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/data/ResourceType.java
new file mode 100644
index 0000000..2fd457a
--- /dev/null
+++ b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/data/ResourceType.java
@@ -0,0 +1,66 @@
+/**
+ * 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.examples.machinedata.data;
+
+import java.util.Map;
+
+import com.google.common.collect.Maps;
+
+/**
+ * This class captures the resources whose usage is collected for each device
+ * <p>ResourceType class.</p>
+ *
+ * @since 0.3.5
+ */
+public enum ResourceType
+{
+
+  CPU("cpu"), RAM("ram"), HDD("hdd");
+
+  private static Map<String, ResourceType> descToResource = Maps.newHashMap();
+
+  static {
+    for (ResourceType type : ResourceType.values()) {
+      descToResource.put(type.desc, type);
+    }
+  }
+
+  private String desc;
+
+  private ResourceType(String desc)
+  {
+    this.desc = desc;
+  }
+
+  @Override
+  public String toString()
+  {
+    return desc;
+  }
+
+  /**
+   * This method returns ResourceType for the given description
+   * @param desc the description
+   * @return
+   */
+  public static ResourceType getResourceTypeOf(String desc)
+  {
+    return descToResource.get(desc);
+  }
+}

http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/operator/CalculatorOperator.java
----------------------------------------------------------------------
diff --git a/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/operator/CalculatorOperator.java b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/operator/CalculatorOperator.java
new file mode 100644
index 0000000..34a9514
--- /dev/null
+++ b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/operator/CalculatorOperator.java
@@ -0,0 +1,279 @@
+/**
+ * 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.examples.machinedata.operator;
+
+import java.io.Serializable;
+import java.math.BigDecimal;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import javax.validation.constraints.Max;
+import javax.validation.constraints.Min;
+
+import org.apache.apex.examples.machinedata.data.MachineInfo;
+import org.apache.apex.examples.machinedata.data.MachineKey;
+import org.apache.apex.examples.machinedata.data.ResourceType;
+import org.apache.apex.examples.machinedata.util.DataTable;
+
+import com.google.common.base.Objects;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.StreamCodec;
+import com.datatorrent.common.util.BaseOperator;
+
+import com.datatorrent.lib.codec.KryoSerializableStreamCodec;
+import com.datatorrent.lib.util.KeyValPair;
+
+/**
+ * <p>
+ * CalculatorOperator class.
+ * </p>
+ *
+ * @since 0.3.5
+ */
+public class CalculatorOperator extends BaseOperator
+{
+
+  private final DataTable<MachineKey, ResourceType, List<Integer>> data = new DataTable<>();
+
+  @Min(1)
+  @Max(99)
+  private int kthPercentile = 95; // kth percentile
+  private boolean computePercentile;
+  private boolean computeSD;
+  private boolean computeMax;
+
+  private int percentileThreshold = 80;
+  private int sdThreshold = 70;
+  private int maxThreshold = 99;
+
+  public final transient DefaultInputPort<MachineInfo> dataPort = new DefaultInputPort<MachineInfo>()
+  {
+    @Override
+    public void process(MachineInfo tuple)
+    {
+      addDataToCache(tuple);
+    }
+
+    /**
+     * Stream codec used for partitioning.
+     */
+    @Override
+    public StreamCodec<MachineInfo> getStreamCodec()
+    {
+      return new MachineInfoStreamCodec();
+    }
+  };
+
+  public final transient DefaultOutputPort<KeyValPair<MachineKey, Map<ResourceType, Double>>> percentileOutputPort = new DefaultOutputPort<>();
+
+  public final transient DefaultOutputPort<KeyValPair<MachineKey, Map<ResourceType, Double>>> sdOutputPort = new DefaultOutputPort<>();
+
+  public final transient DefaultOutputPort<KeyValPair<MachineKey, Map<ResourceType, Integer>>> maxOutputPort = new DefaultOutputPort<>();
+
+  public transient DefaultOutputPort<String> smtpAlert = new DefaultOutputPort<>();
+
+  private void addDataToCache(MachineInfo tuple)
+  {
+    MachineKey machineKey = tuple.getMachineKey();
+    if (!data.containsRow(machineKey)) {
+      data.put(machineKey, ResourceType.CPU, Lists.<Integer>newArrayList());
+      data.put(machineKey, ResourceType.RAM, Lists.<Integer>newArrayList());
+      data.put(machineKey, ResourceType.HDD, Lists.<Integer>newArrayList());
+    }
+    data.get(machineKey, ResourceType.CPU).add(tuple.getCpu());
+    data.get(machineKey, ResourceType.RAM).add(tuple.getRam());
+    data.get(machineKey, ResourceType.HDD).add(tuple.getHdd());
+  }
+
+  @Override
+  public void endWindow()
+  {
+
+    if (computePercentile) {
+      for (MachineKey machineKey : data.rowKeySet()) {
+        Collections.sort(data.get(machineKey, ResourceType.CPU));
+        Collections.sort(data.get(machineKey, ResourceType.RAM));
+        Collections.sort(data.get(machineKey, ResourceType.HDD));
+
+        Map<ResourceType, Double> percentileData = Maps.newHashMap();
+        percentileData.put(ResourceType.CPU, getKthPercentile(data.get(machineKey, ResourceType.CPU)));
+        percentileData.put(ResourceType.RAM, getKthPercentile(data.get(machineKey, ResourceType.RAM)));
+        percentileData.put(ResourceType.HDD, getKthPercentile(data.get(machineKey, ResourceType.HDD)));
+        percentileOutputPort.emit(new KeyValPair<>(machineKey, percentileData));
+
+        for (ResourceType resourceType : percentileData.keySet()) {
+          double percentileValue = percentileData.get(resourceType);
+          if (percentileValue > percentileThreshold) {
+            emitAlert(resourceType, machineKey, percentileValue, "Percentile");
+          }
+        }
+      }
+    }
+    if (computeSD) {
+      for (MachineKey machineKey : data.rowKeySet()) {
+
+        Map<ResourceType, Double> sdData = Maps.newHashMap();
+
+        for (ResourceType resourceType : ResourceType.values()) {
+          sdData.put(resourceType, getSD(data.get(machineKey, resourceType)));
+        }
+        sdOutputPort.emit(new KeyValPair<>(machineKey, sdData));
+
+        for (ResourceType resourceType : sdData.keySet()) {
+          double sdValue = sdData.get(resourceType);
+          if (sdValue > sdThreshold) {
+            emitAlert(resourceType, machineKey, sdValue, "SD");
+          }
+        }
+      }
+    }
+    if (computeMax) {
+      for (MachineKey machineKey : data.rowKeySet()) {
+
+        Map<ResourceType, Integer> maxData = Maps.newHashMap();
+        maxData.put(ResourceType.CPU, Collections.max(data.get(machineKey, ResourceType.CPU)));
+        maxData.put(ResourceType.RAM, Collections.max(data.get(machineKey, ResourceType.RAM)));
+        maxData.put(ResourceType.HDD, Collections.max(data.get(machineKey, ResourceType.HDD)));
+
+        maxOutputPort.emit(new KeyValPair<>(machineKey, maxData));
+
+        for (ResourceType resourceType : maxData.keySet()) {
+          double sdValue = maxData.get(resourceType).doubleValue();
+          if (sdValue > maxThreshold) {
+            emitAlert(resourceType, machineKey, sdValue, "Max");
+          }
+        }
+      }
+    }
+    data.clear();
+  }
+
+  private void emitAlert(ResourceType type, MachineKey machineKey, double alertVal, String prefix)
+  {
+    BigDecimal decimalVal = new BigDecimal(alertVal);
+    decimalVal = decimalVal.setScale(2, BigDecimal.ROUND_HALF_UP);
+    String alertTime = machineKey.getDay() + machineKey.getTimeKey();
+    smtpAlert.emit(prefix + "-" + type.toString().toUpperCase() + " alert at " + alertTime + " " + type + " usage breached current usage: " + decimalVal.doubleValue() + "% threshold: " + percentileThreshold + "%\n\n" + machineKey);
+  }
+
+  private double getKthPercentile(List<Integer> sorted)
+  {
+
+    double val = (kthPercentile * sorted.size()) / 100.0;
+    if (val == (int)val) {
+      // Whole number
+      int idx = (int)val - 1;
+      return (sorted.get(idx) + sorted.get(idx + 1)) / 2.0;
+    } else {
+      int idx = (int)Math.round(val) - 1;
+      return sorted.get(idx);
+    }
+  }
+
+  private double getSD(List<Integer> data)
+  {
+    int sum = 0;
+    for (int i : data) {
+      sum += i;
+    }
+    double avg = sum / (data.size() * 1.0);
+    double sd = 0;
+    for (Integer point : data) {
+      sd += Math.pow(point - avg, 2);
+    }
+    return Math.sqrt(sd);
+  }
+
+  /**
+   * @param kVal the percentile which will be emitted by this operator
+   */
+  public void setKthPercentile(int kVal)
+  {
+    this.kthPercentile = kVal;
+  }
+
+  /**
+   * @param doCompute when true percentile will be computed
+   */
+  public void setComputePercentile(boolean doCompute)
+  {
+    this.computePercentile = doCompute;
+  }
+
+  /**
+   * @param doCompute when true standard deviation will be computed
+   */
+  public void setComputeSD(boolean doCompute)
+  {
+    this.computeSD = doCompute;
+  }
+
+  /**
+   * @param doCompute when true max will be computed
+   */
+  public void setComputeMax(boolean doCompute)
+  {
+    this.computeMax = doCompute;
+  }
+
+  /**
+   * @param threshold for percentile when breached will cause alert
+   */
+  public void setPercentileThreshold(int threshold)
+  {
+    this.percentileThreshold = threshold;
+  }
+
+  /**
+   * @param threshold for standard deviation when breached will cause alert
+   */
+  public void setSDThreshold(int threshold)
+  {
+    this.sdThreshold = threshold;
+  }
+
+  /**
+   * @param threshold for Max when breached will cause alert
+   */
+  public void setMaxThreshold(int threshold)
+  {
+    this.maxThreshold = threshold;
+  }
+
+  public static class MachineInfoStreamCodec extends KryoSerializableStreamCodec<MachineInfo> implements Serializable
+  {
+    public MachineInfoStreamCodec()
+    {
+      super();
+    }
+
+    @Override
+    public int getPartition(MachineInfo o)
+    {
+      return Objects.hashCode(o.getMachineKey().getCustomer(), o.getMachineKey().getOs(), o.getMachineKey().getProduct(), o.getMachineKey().getSoftware1(), o.getMachineKey().getSoftware2(), o.getMachineKey().getSoftware3());
+    }
+
+    private static final long serialVersionUID = 201411031403L;
+  }
+}

http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/operator/MachineInfoAveragingOperator.java
----------------------------------------------------------------------
diff --git a/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/operator/MachineInfoAveragingOperator.java b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/operator/MachineInfoAveragingOperator.java
new file mode 100644
index 0000000..1de676f
--- /dev/null
+++ b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/operator/MachineInfoAveragingOperator.java
@@ -0,0 +1,216 @@
+/**
+ * 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.examples.machinedata.operator;
+
+import java.math.BigDecimal;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.apex.examples.machinedata.data.AverageData;
+import org.apache.apex.examples.machinedata.data.MachineInfo;
+import org.apache.apex.examples.machinedata.data.MachineKey;
+
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.common.util.BaseOperator;
+
+import com.datatorrent.lib.util.KeyHashValPair;
+import com.datatorrent.lib.util.KeyValPair;
+
+/**
+ * This class calculates the average for various resources across different devices for a given key
+ * <p>MachineInfoAveragingOperator class.</p>
+ *
+ * @since 0.9.0
+ */
+@SuppressWarnings("unused")
+public class MachineInfoAveragingOperator extends BaseOperator
+{
+
+  public static final String CPU = "cpu";
+  public static final String RAM = "ram";
+  public static final String HDD = "hdd";
+  public static final String DAY = "day";
+
+  private final transient Map<MachineKey, AverageData> dataMap = new HashMap<>();
+
+  public final transient DefaultOutputPort<KeyValPair<MachineKey, Map<String, String>>> outputPort = new DefaultOutputPort<>();
+
+  public transient DefaultOutputPort<String> smtpAlert = new DefaultOutputPort<>();
+
+  private int threshold = 95;
+
+  /**
+   * Buffer all the tuples as is till end window gets called
+   */
+  public final transient DefaultInputPort<KeyHashValPair<MachineKey, AverageData>> inputPort = new DefaultInputPort<KeyHashValPair<MachineKey, AverageData>>()
+  {
+
+    @Override
+    public void process(KeyHashValPair<MachineKey, AverageData> tuple)
+    {
+      addTuple(tuple);
+    }
+  };
+
+  /**
+   * This method returns the threshold value
+   *
+   * @return
+   */
+  public int getThreshold()
+  {
+    return threshold;
+  }
+
+  /**
+   * This method sets the threshold value. If the average usage for any Resource is above this for a given key, then the alert is sent
+   *
+   * @param threshold the threshold value
+   */
+  public void setThreshold(int threshold)
+  {
+    this.threshold = threshold;
+  }
+
+  /**
+   * This adds the given tuple to the dataMap
+   *
+   * @param tuple input tuple
+   */
+  private void addTuple(KeyHashValPair<MachineKey, AverageData> tuple)
+  {
+    MachineKey key = tuple.getKey();
+    dataMap.put(key, tuple.getValue());
+  }
+
+  @Override
+  public void endWindow()
+  {
+    for (Map.Entry<MachineKey, AverageData> entry : dataMap.entrySet()) {
+      MachineKey key = entry.getKey();
+      AverageData averageResultMap = entry.getValue();
+      Map<String, String> averageResult = Maps.newHashMap();
+      long count = averageResultMap.getCount();
+      double average = averageResultMap.getCpu() / count;
+      averageResult.put(CPU, average + "");
+      emitAlert(average, CPU, key);
+      average = averageResultMap.getHdd() / count;
+      averageResult.put(HDD, average + "");
+      emitAlert(average, HDD, key);
+      average = averageResultMap.getRam() / count;
+      averageResult.put(RAM, average + "");
+      emitAlert(average, RAM, key);
+      averageResult.put(DAY, key.getDay());
+      outputPort.emit(new KeyValPair<>(key, averageResult));
+    }
+    dataMap.clear();
+  }
+
+  private void emitAlert(double average, String resourceType, MachineKey key)
+  {
+    if (average > threshold) {
+      BigDecimal bd = new BigDecimal(average);
+      bd = bd.setScale(2, BigDecimal.ROUND_HALF_UP);
+      String stime = key.getDay() + key.getTimeKey();
+      String skey = getKeyInfo(key);
+      smtpAlert.emit(resourceType.toUpperCase() + " alert at " + stime + " " + resourceType + " usage breached current usage: " + bd.doubleValue() + "% threshold: " + threshold + "%\n\n" + skey);
+    }
+  }
+
+  /**
+   * This method is used to artificially generate alerts
+   *
+   * @param genAlert
+   */
+  public void setGenAlert(boolean genAlert)
+  {
+    Calendar calendar = Calendar.getInstance();
+    long timestamp = System.currentTimeMillis();
+    calendar.setTimeInMillis(timestamp);
+    DateFormat minuteDateFormat = new SimpleDateFormat("HHmm");
+    Date date = calendar.getTime();
+    String timeKey = minuteDateFormat.format(date);
+    DateFormat dayDateFormat = new SimpleDateFormat("dd");
+    String day = dayDateFormat.format(date);
+
+    MachineKey alertKey = new MachineKey(timeKey, day);
+    alertKey.setCustomer(1);
+    alertKey.setProduct(5);
+    alertKey.setOs(10);
+    alertKey.setSoftware1(12);
+    alertKey.setSoftware2(14);
+    alertKey.setSoftware3(6);
+
+    MachineInfo machineInfo = new MachineInfo();
+    machineInfo.setMachineKey(alertKey);
+    machineInfo.setCpu(threshold + 1);
+    machineInfo.setRam(threshold + 1);
+    machineInfo.setHdd(threshold + 1);
+
+    smtpAlert.emit("CPU Alert: CPU Usage threshold (" + threshold + ") breached: current % usage: " + getKeyInfo(alertKey));
+    smtpAlert.emit("RAM Alert: RAM Usage threshold (" + threshold + ") breached: current % usage: " + getKeyInfo(alertKey));
+    smtpAlert.emit("HDD Alert: HDD Usage threshold (" + threshold + ") breached: current % usage: " + getKeyInfo(alertKey));
+  }
+
+  /**
+   * This method returns the String for a given MachineKey instance
+   *
+   * @param key MachineKey instance that needs to be converted to string
+   * @return
+   */
+  private String getKeyInfo(MachineKey key)
+  {
+    StringBuilder sb = new StringBuilder();
+    if (key instanceof MachineKey) {
+      MachineKey mkey = (MachineKey)key;
+      Integer customer = mkey.getCustomer();
+      if (customer != null) {
+        sb.append("customer: " + customer + "\n");
+      }
+      Integer product = mkey.getProduct();
+      if (product != null) {
+        sb.append("product version: " + product + "\n");
+      }
+      Integer os = mkey.getOs();
+      if (os != null) {
+        sb.append("os version: " + os + "\n");
+      }
+      Integer software1 = mkey.getSoftware1();
+      if (software1 != null) {
+        sb.append("software1 version: " + software1 + "\n");
+      }
+      Integer software2 = mkey.getSoftware2();
+      if (software2 != null) {
+        sb.append("software2 version: " + software2 + "\n");
+      }
+      Integer software3 = mkey.getSoftware3();
+      if (software3 != null) {
+        sb.append("software3 version: " + software3 + "\n");
+      }
+    }
+    return sb.toString();
+  }
+}

http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/operator/MachineInfoAveragingPrerequisitesOperator.java
----------------------------------------------------------------------
diff --git a/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/operator/MachineInfoAveragingPrerequisitesOperator.java b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/operator/MachineInfoAveragingPrerequisitesOperator.java
new file mode 100644
index 0000000..14c8d25
--- /dev/null
+++ b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/operator/MachineInfoAveragingPrerequisitesOperator.java
@@ -0,0 +1,88 @@
+/**
+ * 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.examples.machinedata.operator;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.apex.examples.machinedata.data.AverageData;
+import org.apache.apex.examples.machinedata.data.MachineInfo;
+import org.apache.apex.examples.machinedata.data.MachineKey;
+
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.util.KeyHashValPair;
+
+/**
+ * This class calculates the partial sum and count for tuples generated by upstream Operator
+ * <p> MachineInfoAveragingPrerequisitesOperator class. </p>
+ *
+ * @since 0.3.5
+ */
+public class MachineInfoAveragingPrerequisitesOperator extends BaseOperator
+{
+
+  // Aggregate sum of all values seen for a key.
+  private Map<MachineKey, AverageData> sums = new HashMap<>();
+
+  public final transient DefaultOutputPort<KeyHashValPair<MachineKey, AverageData>> outputPort = new DefaultOutputPort<KeyHashValPair<MachineKey, AverageData>>()
+  {
+    @Override
+    public Unifier<KeyHashValPair<MachineKey, AverageData>> getUnifier()
+    {
+      MachineInfoAveragingUnifier unifier = new MachineInfoAveragingUnifier();
+      return unifier;
+    }
+  };
+
+  public transient DefaultInputPort<MachineInfo> inputPort = new DefaultInputPort<MachineInfo>()
+  {
+
+    @Override
+    public void process(MachineInfo tuple)
+    {
+      MachineKey key = tuple.getMachineKey();
+      AverageData averageData = sums.get(key);
+      if (averageData == null) {
+        averageData = new AverageData(tuple.getCpu(), tuple.getHdd(), tuple.getRam(), 1);
+        sums.put(key, averageData);
+      } else {
+        averageData.setCpu(averageData.getCpu() + tuple.getCpu());
+        averageData.setRam(averageData.getRam() + tuple.getRam());
+        averageData.setHdd(averageData.getHdd() + tuple.getHdd());
+        averageData.setCount(averageData.getCount() + 1);
+      }
+    }
+  };
+
+  @Override
+  public void endWindow()
+  {
+
+    for (Map.Entry<MachineKey, AverageData> entry : sums.entrySet()) {
+      if (outputPort.isConnected()) {
+        outputPort.emit(new KeyHashValPair<>(entry.getKey(), entry.getValue()));
+      }
+    }
+    sums.clear();
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/operator/MachineInfoAveragingUnifier.java
----------------------------------------------------------------------
diff --git a/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/operator/MachineInfoAveragingUnifier.java b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/operator/MachineInfoAveragingUnifier.java
new file mode 100644
index 0000000..02aabe7
--- /dev/null
+++ b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/operator/MachineInfoAveragingUnifier.java
@@ -0,0 +1,92 @@
+/**
+ * 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.examples.machinedata.operator;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.apex.examples.machinedata.data.AverageData;
+import org.apache.apex.examples.machinedata.data.MachineKey;
+
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator.Unifier;
+
+import com.datatorrent.lib.util.KeyHashValPair;
+
+/**
+ * This class calculates the partial sum and count for a given key
+ * <p>MachineInfoAveragingUnifier class.</p>
+ *
+ * @since 0.9.0
+ */
+public class MachineInfoAveragingUnifier implements Unifier<KeyHashValPair<MachineKey, AverageData>>
+{
+
+  private Map<MachineKey, AverageData> sums = new HashMap<>();
+  public final transient DefaultOutputPort<KeyHashValPair<MachineKey, AverageData>> outputPort = new DefaultOutputPort<>();
+
+  @Override
+  public void beginWindow(long arg0)
+  {
+    // TODO Auto-generated method stub
+
+  }
+
+  @Override
+  public void endWindow()
+  {
+    for (Map.Entry<MachineKey, AverageData> entry : sums.entrySet()) {
+      outputPort.emit(new KeyHashValPair<>(entry.getKey(), entry.getValue()));
+    }
+    sums.clear();
+
+  }
+
+  @Override
+  public void setup(OperatorContext arg0)
+  {
+    // TODO Auto-generated method stub
+
+  }
+
+  @Override
+  public void teardown()
+  {
+    // TODO Auto-generated method stub
+
+  }
+
+  @Override
+  public void process(KeyHashValPair<MachineKey, AverageData> arg0)
+  {
+    MachineKey tupleKey = arg0.getKey();
+    AverageData averageData = sums.get(tupleKey);
+    AverageData tupleValue = arg0.getValue();
+    if (averageData == null) {
+      sums.put(tupleKey, tupleValue);
+    } else {
+      averageData.setCpu(averageData.getCpu() + tupleValue.getCpu());
+      averageData.setRam(averageData.getRam() + tupleValue.getRam());
+      averageData.setHdd(averageData.getHdd() + tupleValue.getHdd());
+      averageData.setCount(averageData.getCount() + tupleValue.getCount());
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/util/Combinatorics.java
----------------------------------------------------------------------
diff --git a/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/util/Combinatorics.java b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/util/Combinatorics.java
new file mode 100644
index 0000000..0ff8985
--- /dev/null
+++ b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/util/Combinatorics.java
@@ -0,0 +1,101 @@
+/**
+ * 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.examples.machinedata.util;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Generate combinations of elements for the given array of elements.
+ *
+ * Implements nCr = n! / (r! * (n-r)!)
+ *
+ * @since 0.3.5
+ */
+public class Combinatorics<T>
+{
+
+  private T[] values;
+  private int size = -1;
+  private List<T> result;
+  private Map<Integer, List<T>> resultMap = new HashMap<Integer, List<T>>();
+  private int resultMapSize = 0;
+
+  /**
+   * Generates all possible combinations with all the sizes.
+   *
+   * @param values
+   */
+  public Combinatorics(T[] values)
+  {
+    this.values = values;
+    this.size = -1;
+    this.result = new ArrayList<>();
+  }
+
+  /**
+   * Generates all possible combinations with the given size.
+   *
+   * @param values
+   * @param size
+   */
+  public Combinatorics(T[] values, int size)
+  {
+    this.values = values;
+    this.size = size;
+    this.result = new ArrayList<>();
+  }
+
+  public Map<Integer, List<T>> generate()
+  {
+
+    if (size == -1) {
+      size = values.length;
+      for (int i = 1; i <= size; i++) {
+        int[] tmp = new int[i];
+        Arrays.fill(tmp, -1);
+        generateCombinations(0, 0, tmp);
+      }
+    } else {
+      int[] tmp = new int[size];
+      Arrays.fill(tmp, -1);
+      generateCombinations(0, 0, tmp);
+    }
+    return resultMap;
+  }
+
+  public void generateCombinations(int start, int depth, int[] tmp)
+  {
+    if (depth == tmp.length) {
+      for (int j = 0; j < depth; j++) {
+        result.add(values[tmp[j]]);
+      }
+      resultMap.put(++resultMapSize, result);
+      result = new ArrayList<>();
+      return;
+    }
+    for (int i = start; i < values.length; i++) {
+      tmp[depth] = i;
+      generateCombinations(i + 1, depth + 1, tmp);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/util/DataTable.java
----------------------------------------------------------------------
diff --git a/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/util/DataTable.java b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/util/DataTable.java
new file mode 100644
index 0000000..a147268
--- /dev/null
+++ b/examples/machinedata/src/main/java/org/apache/apex/examples/machinedata/util/DataTable.java
@@ -0,0 +1,75 @@
+/**
+ * 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.examples.machinedata.util;
+
+import java.util.Map;
+import java.util.Set;
+
+import javax.annotation.Nullable;
+
+import com.google.common.collect.Maps;
+
+/**
+ * <p>DataTable class.</p>
+ *
+ * @since 0.3.5
+ */
+public class DataTable<R,C,E>
+{
+
+  //machineKey, [cpu,ram,hdd] -> value
+  private final Map<R,Map<C,E>> table = Maps.newHashMap();
+
+  public boolean containsRow(R rowKey)
+  {
+    return table.containsKey(rowKey);
+  }
+
+  public void put(R rowKey,C colKey, E entry)
+  {
+    if (!containsRow(rowKey)) {
+      table.put(rowKey, Maps.<C,E>newHashMap());
+    }
+    table.get(rowKey).put(colKey, entry);
+  }
+
+  @Nullable
+  public E get(R rowKey, C colKey)
+  {
+    if (!containsRow(rowKey)) {
+      return null;
+    }
+    return table.get(rowKey).get(colKey);
+  }
+
+  public Set<R> rowKeySet()
+  {
+    return table.keySet();
+  }
+
+  public void clear()
+  {
+    table.clear();
+  }
+
+  public Map<R,Map<C,E>> getTable()
+  {
+    return table;
+  }
+}

http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/machinedata/src/main/resources/META-INF/properties.xml
----------------------------------------------------------------------
diff --git a/examples/machinedata/src/main/resources/META-INF/properties.xml b/examples/machinedata/src/main/resources/META-INF/properties.xml
new file mode 100644
index 0000000..4ceb6b9
--- /dev/null
+++ b/examples/machinedata/src/main/resources/META-INF/properties.xml
@@ -0,0 +1,139 @@
+<!--
+
+    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.
+
+-->
+<configuration>
+  <property>
+    <name>dt.attr.STREAMING_WINDOW_SIZE_MILLIS</name>
+    <value>1000</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Receiver.attr.APPLICATION_WINDOW_COUNT
+    </name>
+    <value>5</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Receiver.attr.PARTITIONER</name>
+    <value>com.datatorrent.common.partitioner.StatelessPartitioner:1</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.DimensionsGenerator.attr.APPLICATION_WINDOW_COUNT
+    </name>
+    <value>5</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.stream.Events.locality
+    </name>
+    <value>CONTAINER_LOCAL</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.DimensionsGenerator.inputport.inputPort.attr.PARTITION_PARALLEL
+    </name>
+    <value>true</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Aggregator.inputport.inputPort.attr.PARTITION_PARALLEL
+    </name>
+    <value>true</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Aggregator.attr.APPLICATION_WINDOW_COUNT
+    </name>
+    <value>5</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.AverageCalculator.attr.APPLICATION_WINDOW_COUNT
+    </name>
+    <value>5</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Persister.inputport.input.attr.PARTITION_PARALLEL
+    </name>
+    <value>true</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Persister.attr.APPLICATION_WINDOW_COUNT
+    </name>
+    <value>5</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Persister.store.dbIndex
+    </name>
+    <value>2</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Persister.store.host
+    </name>
+    <value></value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Persister.store.port
+    </name>
+    <value>6379</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.port.*.attr.QUEUE_CAPACITY</name>
+    <value>32000</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Alerter.from
+    </name>
+    <value></value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Alerter.subject
+    </name>
+    <value>Alert!!!</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Alerter.recipients.TO
+    </name>
+    <value></value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Alerter.content
+    </name>
+    <value>{}</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Alerter.smtpHost
+    </name>
+    <value>localhost</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Alerter.smtpPort
+    </name>
+    <value>25</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Alerter.useSsl
+    </name>
+    <value>false</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.operator.Aggregator.outputport.outputPort.attr.UNIFIER_LIMIT
+    </name>
+    <value>8</value>
+  </property>
+  <property>
+    <name>dt.application.MachineDataExample.stream.DimensionalData.locality
+    </name>
+    <value>THREAD_LOCAL</value>
+  </property>
+</configuration>

http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/machinedata/src/test/java/org/apache/apex/examples/machinedata/CalculatorOperatorTest.java
----------------------------------------------------------------------
diff --git a/examples/machinedata/src/test/java/org/apache/apex/examples/machinedata/CalculatorOperatorTest.java b/examples/machinedata/src/test/java/org/apache/apex/examples/machinedata/CalculatorOperatorTest.java
new file mode 100644
index 0000000..c69a2aa
--- /dev/null
+++ b/examples/machinedata/src/test/java/org/apache/apex/examples/machinedata/CalculatorOperatorTest.java
@@ -0,0 +1,197 @@
+/**
+ * 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.examples.machinedata;
+
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.apex.examples.machinedata.data.MachineInfo;
+import org.apache.apex.examples.machinedata.data.MachineKey;
+import org.apache.apex.examples.machinedata.data.ResourceType;
+import org.apache.apex.examples.machinedata.operator.CalculatorOperator;
+
+
+import com.google.common.collect.ImmutableList;
+
+import com.datatorrent.lib.testbench.CollectorTestSink;
+import com.datatorrent.lib.util.KeyValPair;
+import com.datatorrent.lib.util.TimeBucketKey;
+
+/**
+ * @since 0.3.5
+ */
+public class CalculatorOperatorTest
+{
+  private static DateFormat minuteDateFormat = new SimpleDateFormat("HHmm");
+  private static Logger LOG = LoggerFactory.getLogger(CalculatorOperatorTest.class);
+
+  /**
+   * Test node logic emits correct results
+   */
+  @Test
+  public void testNodeProcessing() throws Exception
+  {
+    CalculatorOperator calculatorOperator = new CalculatorOperator();
+    calculatorOperator.setup(null);
+
+    calculatorOperator.setComputePercentile(true);
+    calculatorOperator.setComputeMax(true);
+    calculatorOperator.setComputeSD(true);
+
+    testPercentile(calculatorOperator);
+  }
+
+  public void testPercentile(CalculatorOperator oper)
+  {
+
+    CollectorTestSink sortSink = new CollectorTestSink();
+    oper.percentileOutputPort.setSink(sortSink);
+    oper.setKthPercentile(50);
+    Calendar calendar = Calendar.getInstance();
+    Date date = calendar.getTime();
+    String timeKey = minuteDateFormat.format(date);
+    String day = calendar.get(Calendar.DAY_OF_MONTH) + "";
+
+    Integer vs = new Integer(1);
+    MachineKey mk = new MachineKey(timeKey, day, vs, vs, vs, vs, vs, vs, vs);
+
+    oper.beginWindow(0);
+
+    MachineInfo info = new MachineInfo(mk, 1, 1, 1);
+    oper.dataPort.process(info);
+
+    info.setCpu(2);
+    oper.dataPort.process(info);
+
+    info.setCpu(3);
+    oper.dataPort.process(info);
+
+    oper.endWindow();
+
+    Assert.assertEquals("number emitted tuples", 1, sortSink.collectedTuples.size());
+    for (Object o : sortSink.collectedTuples) {
+      LOG.debug(o.toString());
+      KeyValPair<TimeBucketKey, Map<ResourceType, Double>> keyValPair = (KeyValPair<TimeBucketKey, Map<ResourceType, Double>>)o;
+      Assert.assertEquals("emitted value for 'cpu' was ", 2.0, keyValPair.getValue().get(ResourceType.CPU), 0);
+      Assert.assertEquals("emitted value for 'hdd' was ", 1.0, keyValPair.getValue().get(ResourceType.HDD), 0);
+      Assert.assertEquals("emitted value for 'ram' was ", 1.0, keyValPair.getValue().get(ResourceType.RAM), 0);
+
+    }
+    LOG.debug("Done percentile testing\n");
+
+  }
+
+  public void testStandarDeviation(CalculatorOperator oper)
+  {
+    CollectorTestSink sortSink = new CollectorTestSink();
+    oper.sdOutputPort.setSink(sortSink);
+    Calendar calendar = Calendar.getInstance();
+    Date date = calendar.getTime();
+    String timeKey = minuteDateFormat.format(date);
+    String day = calendar.get(Calendar.DAY_OF_MONTH) + "";
+
+    Integer vs = new Integer(1);
+    MachineKey mk = new MachineKey(timeKey, day, vs, vs, vs, vs, vs, vs, vs);
+
+    oper.beginWindow(0);
+
+    MachineInfo info = new MachineInfo(mk, 1, 1, 1);
+    oper.dataPort.process(info);
+
+    info.setCpu(2);
+    oper.dataPort.process(info);
+
+    info.setCpu(3);
+    oper.dataPort.process(info);
+
+    oper.endWindow();
+
+    Assert.assertEquals("number emitted tuples", 1, sortSink.collectedTuples.size());
+    for (Object o : sortSink.collectedTuples) {
+      LOG.debug(o.toString());
+      KeyValPair<TimeBucketKey, Map<ResourceType, Double>> keyValPair = (KeyValPair<TimeBucketKey, Map<ResourceType, Double>>)o;
+      Assert.assertEquals("emitted value for 'cpu' was ", getSD(ImmutableList.of(1, 2, 3)), keyValPair.getValue().get(ResourceType.CPU), 0);
+      Assert.assertEquals("emitted value for 'hdd' was ", getSD(ImmutableList.of(1, 1, 1)), keyValPair.getValue().get(ResourceType.HDD), 0);
+      Assert.assertEquals("emitted value for 'ram' was ", getSD(ImmutableList.of(1, 1, 1)), keyValPair.getValue().get(ResourceType.RAM), 0);
+
+    }
+    LOG.debug("Done sd testing\n");
+
+  }
+
+  private final double getSD(List<Integer> input)
+  {
+    int sum = 0;
+    for (int i : input) {
+      sum += i;
+    }
+    double avg = sum / (input.size() * 1.0);
+    double sd = 0;
+    for (Integer point : input) {
+      sd += Math.pow(point - avg, 2);
+    }
+    return Math.sqrt(sd);
+  }
+
+  public void testMax(CalculatorOperator oper)
+  {
+    CollectorTestSink sortSink = new CollectorTestSink();
+    oper.maxOutputPort.setSink(sortSink);
+    Calendar calendar = Calendar.getInstance();
+    Date date = calendar.getTime();
+    String timeKey = minuteDateFormat.format(date);
+    String day = calendar.get(Calendar.DAY_OF_MONTH) + "";
+
+    Integer vs = new Integer(1);
+    MachineKey mk = new MachineKey(timeKey, day, vs, vs, vs, vs, vs, vs, vs);
+
+    oper.beginWindow(0);
+
+    MachineInfo info = new MachineInfo(mk, 1, 1, 1);
+    oper.dataPort.process(info);
+
+    info.setCpu(2);
+    oper.dataPort.process(info);
+
+    info.setCpu(3);
+    oper.dataPort.process(info);
+
+    oper.endWindow();
+
+    Assert.assertEquals("number emitted tuples", 1, sortSink.collectedTuples.size());
+    for (Object o : sortSink.collectedTuples) {
+      LOG.debug(o.toString());
+      KeyValPair<TimeBucketKey, Map<ResourceType, Double>> keyValPair = (KeyValPair<TimeBucketKey, Map<ResourceType, Double>>)o;
+      Assert.assertEquals("emitted value for 'cpu' was ", 3, keyValPair.getValue().get(ResourceType.CPU), 0);
+      Assert.assertEquals("emitted value for 'hdd' was ", 1, keyValPair.getValue().get(ResourceType.HDD), 0);
+      Assert.assertEquals("emitted value for 'ram' was ", 1, keyValPair.getValue().get(ResourceType.RAM), 0);
+
+    }
+    LOG.debug("Done max testing\n");
+
+  }
+}

http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/machinedata/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/examples/machinedata/src/test/resources/log4j.properties b/examples/machinedata/src/test/resources/log4j.properties
new file mode 100644
index 0000000..cf0d19e
--- /dev/null
+++ b/examples/machinedata/src/test/resources/log4j.properties
@@ -0,0 +1,43 @@
+#
+# 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.
+#
+
+log4j.rootLogger=DEBUG,CONSOLE
+
+log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
+log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
+log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} [%t] %-5p %c{2} %M - %m%n
+log4j.appender.CONSOLE.threshold=${test.log.console.threshold}
+test.log.console.threshold=DEBUG
+
+log4j.appender.RFA=org.apache.log4j.RollingFileAppender
+log4j.appender.RFA.layout=org.apache.log4j.PatternLayout
+log4j.appender.RFA.layout.ConversionPattern=%d{ISO8601} [%t] %-5p %c{2} %M - %m%n
+log4j.appender.RFA.File=/tmp/app.log
+
+# to enable, add SYSLOG to rootLogger
+log4j.appender.SYSLOG=org.apache.log4j.net.SyslogAppender
+log4j.appender.SYSLOG.syslogHost=127.0.0.1
+log4j.appender.SYSLOG.layout=org.apache.log4j.PatternLayout
+log4j.appender.SYSLOG.layout.conversionPattern=${dt.cid} %-5p [%t] %c{2} %x - %m%n
+log4j.appender.SYSLOG.Facility=LOCAL1
+
+log4j.logger.org=info
+#log4j.logger.org.apache.commons.beanutils=warn
+log4j.logger.com.datatorrent=debug
+log4j.logger.org.apache.apex=debug

http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/d5bf96ca/examples/mobile/pom.xml
----------------------------------------------------------------------
diff --git a/examples/mobile/pom.xml b/examples/mobile/pom.xml
new file mode 100644
index 0000000..cb40887
--- /dev/null
+++ b/examples/mobile/pom.xml
@@ -0,0 +1,64 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <artifactId>malhar-examples-mobile</artifactId>
+  <packaging>jar</packaging>
+
+  <name>Apache Apex Malhar Mobile Example</name>
+  <description></description>
+
+  <parent>
+    <groupId>org.apache.apex</groupId>
+    <artifactId>malhar-examples</artifactId>
+    <version>3.7.0-SNAPSHOT</version>
+  </parent>
+
+  <properties>
+    <skipTests>true</skipTests>
+  </properties>
+
+  <dependencies>
+    <!-- add your dependencies here -->
+    <dependency>
+      <groupId>org.eclipse.jetty</groupId>
+      <artifactId>jetty-servlet</artifactId>
+      <version>8.1.10.v20130312</version>
+      <scope>test</scope>
+      <type>jar</type>
+    </dependency>
+    <dependency>
+      <groupId>org.eclipse.jetty</groupId>
+      <artifactId>jetty-websocket</artifactId>
+      <version>8.1.10.v20130312</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-lang3</artifactId>
+      <version>3.1</version>
+      <type>jar</type>
+    </dependency>
+  </dependencies>
+
+</project>


Mime
View raw message