Cleanup, file operator changes

WeatherStreamingApp/src/main/java/com/dt/weather/app/WeatherApp.java

@@ -19,14 +19,10 @@
 import com.datatorrent.api.annotation.ApplicationAnnotation;
 import com.datatorrent.common.partitioner.StatelessPartitioner;
 import com.datatorrent.contrib.kafka.KafkaSinglePortOutputOperator;
-import com.datatorrent.stram.engine.PortContext;
 import com.dt.weather.constants.WeatherConstants;
 import com.dt.weather.converter.OutputConverter;
 import com.dt.weather.counter.KeyValChangeAggregator;
-import com.dt.weather.counter.KeyValChangeAlert;
-import com.dt.weather.event.convertor.SinglePortWeatherEventConvertor;
 import com.dt.weather.input.JSONFileInputOperator;
-import com.dt.weather.input.SimpleFileReader;
 
 @ApplicationAnnotation(name = "WeatherApp")
 public class WeatherApp implements StreamingApplication
@@ -46,11 +42,12 @@ public void populateDAG(DAG dag, Configuration conf)
 
     fileReader.setMatchKey(conf.get(WeatherConstants.MATCH_KEY, "description"));
 
-    //Uncomment the rename logic in simple file reader when the regex works
-  //  fileReader.getScanner().setFilePatternRegexp(".json");
+    fileReader
+        .setProcessedDirPath("/Users/dev/checkout/personalGit/devel2/DTStreaming/WeatherStreamingApp/src/test/resources/data-processed/");
 
     fileReader.setScanIntervalMillis(0);
     fileReader.setEmitBatchSize(1);
+
 
     //Add the overall counter
     KeyValChangeAggregator<String, Integer> counter = dag.addOperator("GlobalCounter",
@@ -74,6 +71,8 @@ public void populateDAG(DAG dag, Configuration conf)
     dag.addStream("InputRecords", fileReader.output, counter.data).setLocality(Locality.CONTAINER_LOCAL);
 
     dag.addStream("Unifier Output", counter.alert, opConv.data);
+
+    dag.addStream("UnifierCumulative Output", counter.sum, opConv.dataCumulative);
 
     dag.addStream("Convert Output", opConv.output, kafkaOutputOperator.inputPort);
 
@@ -131,21 +130,4 @@ public static Configuration readPropertiesFile(String fileName)
     return config;
   }
 
-  public static void main(String[] args) throws Exception
-  {
-
-    LocalMode lma = LocalMode.newInstance();
-    //  Configuration conf = new Configuration(false);
-
-    Configuration conf = new Application()
-        .readPropertiesFile("/Users/dev/workspace/mydtapp/src/test/resources/localmode.properties");
-    // conf.addResource(.getClass().getResourceAsStream("/META-INF/properties.xml"));
-
-    lma.prepareDAG(new Application(), conf);
-    LocalMode.Controller lc = lma.getController();
-    lc.setHeartbeatMonitoringEnabled(false);
-    lc.run(10000);
-
-  }
-
 }

WeatherStreamingApp/src/main/java/com/dt/weather/constants/WeatherConstants.java

@@ -18,6 +18,5 @@ public class WeatherConstants
   public static final String RECORD_SEPARATOR = ",";
   public static final String TUPLE_SEPARATOR = ":";
   public static final String MATCH_KEY = "description";
-
-
+
 }